Multitenancy

This module implements multitenancy, meaning that your application can be used by a number of independent subjects (tenants, for example companies) without interfering with each other.

Getting started

To set up an application with multi-tenant web interface, create an application with a web server and initialize asab.web.tenant.TenantService. Tenant service automatically tries to install tenant context wrapper to your web handlers, which enables you to access the request's tenant context using asab.contextvars.Tenant.get().

import asab
import asab.web
import asab.web.tenant
import asab.contextvars

class MyApplication(asab.Application):
    def __init__(self):
        super().__init__()

        # Initialize web module
        asab.web.create_web_server(self)

        # Initialize tenant service
        self.TenantService = asab.web.tenant.TenantService(self)

Note

If your app has more than one web container, you will need to call TenantService.install(web_container) to apply the tenant context wrapper.

This also adds the requirement for tenant parameter in the URL of every request - either in the path or in the query.

Mandatory tenant in path

If tenant context is mandatory for your endpoint, it is recommended to require the tenant parameter in the URL path, such as:

import asab
import asab.web
import asab.web.tenant
import asab.contextvars

class NotesApplication(asab.Application):
    def __init__(self):
        super().__init__()
        web = asab.web.create_web_server(self)
        tenant_svc = asab.web.tenant.TenantService(self)

        web.add_get("/{tenant}/note", self.list_notes)  # Tenant parameter required in path

    async def list_notes(self, request):
        tenant = asab.contextvars.Tenant.get()
        print("Requesting notes for tenant {!r}...".format(tenant))

Note

It is a good practice to have tenant as the first component of the URL path if possible.

Mandatory tenant in query

When the tenant context is mandatory for your endpoint, but it is not feasible to have the tenant parameter hard-baked into the path, define your endpoint path without the tenant path parameter. The handler with require tenant to be present in the URL query. Requests without the required parameter will result in asab.exceptions.ValidationError (HTTP 400).

import asab
import asab.web
import asab.web.tenant
import asab.contextvars

class NotesApplication(asab.Application):
    def __init__(self):
        super().__init__()
        web = asab.web.create_web_server(self)
        tenant_svc = asab.web.tenant.TenantService(self)

        web.add_get("/note", self.list_notes)  # No tenant parameter in path!

    async def list_notes(self, request):
        tenant = asab.contextvars.Tenant.get()
        print("Requesting notes for tenant {!r}...".format(tenant))

Optional tenant in query

When the tenant context is optional for your endpoint (or when the endpoint does not use tenants at all), define its path without the tenant parameter in path and decorate the method handler with @asab.web.tenant.allow_no_tenant. Requests without the tenant parameter will have their Tenant context set to None.

import asab
import asab.web
import asab.web.tenant
import asab.contextvars

class NotesApplication(asab.Application):
    def __init__(self):
        super().__init__()
        web = asab.web.create_web_server(self)
        tenant_svc = asab.web.tenant.TenantService(self)

        web.add_get("/{tenant}/note", self.list_notes)  # No tenant parameter in path!

    @asab.web.tenant.allow_no_tenant  # Allow requests with undefined tenant!
    async def list_notes(self, request):
        tenant = asab.contextvars.Tenant.get()
        if tenant is None:
            print("Requesting notes without any tenant. Not sure what to do...")
        else:
            print("Requesting notes for tenant {!r}...".format(tenant))

Working with known tenants

When you provide tenant_url or tenant ids in the configuration, TenantService will make the set of known tenants available through its Tenants property. You can also make use of the TenantService.is_tenant_known(tenant) method.

Note

If you only want to use the service to access known tenants and do not need the web middleware, initialize TenantService with set_up_web_wrapper argument set to False.

Configuration

The asab.web.tenant module is configured in the [tenants] section with the following options:

Option	Type	Meaning
ids	List of strings	(Optional) Known tenant IDs.
tenant_url	URL	(Optional) Location of a JSON array of known tenant IDs.

Reference

asab.web.tenant.TenantService

Bases: Service

Provides set of known tenants and tenant extraction for web requests.

Source code in asab/web/tenant/service.py

class TenantService(Service):
	"""
	Provides set of known tenants and tenant extraction for web requests.
	"""

	def __init__(self, app, service_name: str = "asab.TenantService", auto_install_web_wrapper: bool = True):
		"""
		Initialize and register a new TenantService.

		Args:
			app: ASAB application.
			service_name: ASAB service identifier.
			auto_install_web_wrapper: Whether to automatically install tenant context wrapper to WebContainer.
		"""
		super().__init__(app, service_name)
		self.App = app
		self.Providers = []  # Must be a list to be deterministic


		auth_svc = self.App.get_service("asab.AuthService")
		if auth_svc is not None:
			raise RuntimeError("Please initialize TenantService before AuthService.")

		self._prepare_providers()
		if auto_install_web_wrapper:
			self._try_auto_install()


	def _prepare_providers(self):
		if Config.get("tenants", "ids", fallback=None):
			from .providers import StaticTenantProvider
			self.Providers.append(StaticTenantProvider(self.App, Config["tenants"]))

		if Config.get("tenants", "tenant_url", fallback=None):
			from .providers import WebTenantProvider
			self.Providers.append(WebTenantProvider(self.App, Config["tenants"]))


	async def initialize(self, app):
		if len(self.Providers) == 0:
			L.error(
				"TenantService requires at least one provider. "
				"Specify either `tenant_url` or `ids` in the [tenants] config section."
			)

		for provider in self.Providers:
			await provider.initialize(app)


	@property
	def Tenants(self) -> typing.Set[str]:
		"""
		Get the set of known tenant IDs.

		Returns:
			The set of known tenant IDs.
		"""
		return self.get_tenants()


	def get_tenants(self) -> typing.Set[str]:
		"""
		Get the set of known tenant IDs.

		Returns:
			The set of known tenant IDs.
		"""
		tenants = set()
		for provider in self.Providers:
			tenants |= provider.get_tenants()

		return tenants


	def is_tenant_known(self, tenant: str) -> bool:
		"""
		Check if the tenant is among known tenants.

		Args:
			tenant: Tenant ID to check.

		Returns:
			Whether the tenant is known.
		"""
		if tenant is None:
			return False
		if len(self.Providers) == 0:
			L.warning("No tenant provider registered.")
			return False
		for provider in self.Providers:
			if provider.is_tenant_known(tenant):
				return True
		return False


	def install(self, web_container):
		"""
		Apply tenant context wrappers to all web handlers in the web container.

		Args:
			web_container: Web container to add tenant context to.
		"""
		web_service = self.App.get_service("asab.WebService")

		# Check that the middleware has not been installed yet
		for middleware in web_container.WebApp.on_startup:
			if middleware == self._set_up_tenant_web_wrapper:
				if len(web_service.Containers) == 1:
					raise RuntimeError(
						"WebContainer has tenant middleware installed already. "
						"You don't need to call `TenantService.install()` in applications with a single WebContainer; "
						"it is called automatically at init time."
					)
				else:
					raise RuntimeError("WebContainer has tenant middleware installed already.")

		web_container.WebApp.on_startup.append(self._set_up_tenant_web_wrapper)


	def get_web_wrapper_position(self, web_container) -> typing.Optional[int]:
		"""
		Check if tenant web wrapper is installed in container and where.

		Args:
			web_container: Web container to inspect.

		Returns:
			typing.Optional[int]: The index at which the wrapper is located, or `None` if it is not installed.
		"""
		try:
			return web_container.WebApp.on_startup.index(self._set_up_tenant_web_wrapper)
		except ValueError:
			return None


	def _try_auto_install(self):
		"""
		If there is exactly one web container, install tenant middleware on it.
		"""
		web_service = self.App.get_service("asab.WebService")
		if web_service is None:
			return
		if len(web_service.Containers) != 1:
			return
		web_container = web_service.WebContainer

		self.install(web_container)
		L.debug("WebContainer tenant wrapper will be installed automatically.")


	async def _set_up_tenant_web_wrapper(self, aiohttp_app: aiohttp.web.Application):
		"""
		Inspect all registered handlers and wrap them in decorators according to their parameters.
		"""
		for route in aiohttp_app.router.routes():
			# Skip non-coroutines
			if not inspect.iscoroutinefunction(route.handler):
				continue

			try:
				set_handler_tenant(self, route)
			except Exception as e:
				raise RuntimeError(
					"Failed to initialize tenant context for handler {!r}.".format(route.handler.__qualname__)
				) from e

Tenants: typing.Set[str] property

Get the set of known tenant IDs.

Returns:

Type	Description
Set[str]	The set of known tenant IDs.

__init__(app, service_name='asab.TenantService', auto_install_web_wrapper=True)

Initialize and register a new TenantService.

Parameters:

Name	Type	Description	Default
app		ASAB application.	required
service_name	str	ASAB service identifier.	'asab.TenantService'
auto_install_web_wrapper	bool	Whether to automatically install tenant context wrapper to WebContainer.	True

Source code in asab/web/tenant/service.py

def __init__(self, app, service_name: str = "asab.TenantService", auto_install_web_wrapper: bool = True):
	"""
	Initialize and register a new TenantService.

	Args:
		app: ASAB application.
		service_name: ASAB service identifier.
		auto_install_web_wrapper: Whether to automatically install tenant context wrapper to WebContainer.
	"""
	super().__init__(app, service_name)
	self.App = app
	self.Providers = []  # Must be a list to be deterministic


	auth_svc = self.App.get_service("asab.AuthService")
	if auth_svc is not None:
		raise RuntimeError("Please initialize TenantService before AuthService.")

	self._prepare_providers()
	if auto_install_web_wrapper:
		self._try_auto_install()

get_tenants()

Get the set of known tenant IDs.

Returns:

Type	Description
Set[str]	The set of known tenant IDs.

Source code in asab/web/tenant/service.py

def get_tenants(self) -> typing.Set[str]:
	"""
	Get the set of known tenant IDs.

	Returns:
		The set of known tenant IDs.
	"""
	tenants = set()
	for provider in self.Providers:
		tenants |= provider.get_tenants()

	return tenants

get_web_wrapper_position(web_container)

Check if tenant web wrapper is installed in container and where.

Parameters:

Name	Type	Description	Default
web_container		Web container to inspect.	required

Returns:

Type	Description
Optional[int]	typing.Optional[int]: The index at which the wrapper is located, or None if it is not installed.

Source code in asab/web/tenant/service.py

def get_web_wrapper_position(self, web_container) -> typing.Optional[int]:
	"""
	Check if tenant web wrapper is installed in container and where.

	Args:
		web_container: Web container to inspect.

	Returns:
		typing.Optional[int]: The index at which the wrapper is located, or `None` if it is not installed.
	"""
	try:
		return web_container.WebApp.on_startup.index(self._set_up_tenant_web_wrapper)
	except ValueError:
		return None

install(web_container)

Apply tenant context wrappers to all web handlers in the web container.

Parameters:

Name	Type	Description	Default
web_container		Web container to add tenant context to.	required

Source code in asab/web/tenant/service.py

def install(self, web_container):
	"""
	Apply tenant context wrappers to all web handlers in the web container.

	Args:
		web_container: Web container to add tenant context to.
	"""
	web_service = self.App.get_service("asab.WebService")

	# Check that the middleware has not been installed yet
	for middleware in web_container.WebApp.on_startup:
		if middleware == self._set_up_tenant_web_wrapper:
			if len(web_service.Containers) == 1:
				raise RuntimeError(
					"WebContainer has tenant middleware installed already. "
					"You don't need to call `TenantService.install()` in applications with a single WebContainer; "
					"it is called automatically at init time."
				)
			else:
				raise RuntimeError("WebContainer has tenant middleware installed already.")

	web_container.WebApp.on_startup.append(self._set_up_tenant_web_wrapper)

is_tenant_known(tenant)

Check if the tenant is among known tenants.

Parameters:

Name	Type	Description	Default
tenant	str	Tenant ID to check.	required

Returns:

Type	Description
bool	Whether the tenant is known.

Source code in asab/web/tenant/service.py

def is_tenant_known(self, tenant: str) -> bool:
	"""
	Check if the tenant is among known tenants.

	Args:
		tenant: Tenant ID to check.

	Returns:
		Whether the tenant is known.
	"""
	if tenant is None:
		return False
	if len(self.Providers) == 0:
		L.warning("No tenant provider registered.")
		return False
	for provider in self.Providers:
		if provider.is_tenant_known(tenant):
			return True
	return False

asab.web.tenant.allow_no_tenant(handler)

Allow receiving requests without tenant parameter.

Parameters:

Name	Type	Description	Default
handler		Web handler method	required

Returns:

Type	Description
	Wrapped web handler that allows requests with undefined tenant.

Source code in asab/web/tenant/decorator.py

def allow_no_tenant(handler):
	"""
	Allow receiving requests without tenant parameter.

	Args:
		handler: Web handler method

	Returns:
		Wrapped web handler that allows requests with undefined tenant.
	"""
	handler.AllowNoTenant = True

	@functools.wraps(handler)
	async def _allow_no_tenant_wrapper(*args, **kwargs):
		return await handler(*args, **kwargs)

	return _allow_no_tenant_wrapper