Skip to content

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 Exception("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):
		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
		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 Exception(
						"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 Exception("WebContainer has tenant middleware installed already.")

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


	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.info("WebContainer tenant context 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 Exception(
					"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 Exception("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

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 Exception(
					"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 Exception("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
	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 wrapper(*args, **kwargs):
		return await handler(*args, **kwargs)

	return wrapper