Web server

For starting, accessing, and manipulating of a web server, ASAB provides asab.web module together with asab.WebService and asab.web.WebContainer. This module offers an integration of aiohttp web server. It is possible to run multiple web servers from one application. The configuration for each server is stored in dedicated web container. Web Service registers these containers and runs the servers.

Tip

For a quick start, we recommend reading the official aiohttp tutorial on how to run a simple web server.

In order to use ASAB Web Service, first make sure that you have aiohttp module installed:

python3 -m pip install aiohttp

Handlers, routes, and resources

aiohttp servers use the concept of handlers, routes, and resources. Here we provide a very quick explanation that should help you to get into the terminology.

• Handler, more precisely "a web request handler", is a function that does the logic when you send an HTTP request to the endpoint. It is a coroutine that accepts aiohttp.web.Request instance as its only parameter and returns a aiohttp.web.Response object.

  async def handler(request):
      return aiohttp.web.Response()
  

• Route corresponds to handling HTTP method by calling web handler. Routes are added to the route table that is stored in the Web Application object.

  web_app.add_routes(
      [web.get('/path1', get_1),
      web.post('/path1', post_1),
      web.get('/path2', get_2),
      web.post('/path2', post_2)]
  

• Resource is an entry in route table which corresponds to requested URL. Resource in turn has at least one route. When you add a route, the resource object is created under the hood.

Running a simple web server

Creating a web server 1: One method does it all!

To build a web server quickly, ASAB provides a method asab.web.create_web_server() that does all the magic.

import asab.web
import aiohttp

class MyApplication(asab.Application):
    async def initialize(): #(1)!
        web = asab.web.create_web_server(self) #(2)!
        web.add_get('/hello', self.hello) #(3)!

    # This is the web request handler
    async def hello(self, request): #(4)!
        return aiohttp.web.Response(data="Hello from your ASAB server!\n")

if __name__ == '__main__':
    app = MyApplication()
    app.run()

1. Web server configuration should be prepared during the init-time of the application lifecycle.
2. asab.web.create_web_server() creates web server and returns instance of aiohttp.web.UrlDispatcher object (which is usually referred to as a "router"). The method takes the argument app which is used as the reference to the base asab.Application object, and optional parameters that expand configuration options.
3. You can easily create a new endpoint by aiohttp.web.UrlDispatcher.add_route() method that appends a request handler to the router table. This one is a shortcut for adding a GET handler.
4. A request handler must be a coroutine that accepts aiohttp.web.Request instance as its only parameter and returns aiohttp.web.Response instance.

By default, ASAB server runs on 0.0.0.0:8080. You can test if your server is working by sending a request to the /hello endpoint, e.g., via the curl command:

curl http://localhost:8080/hello

and you should get the response:

Hello from your ASAB server!

Creating a web server 2: Under the hood.

The code below does exactly the same as in the previous example.

import asab.web
import aiohttp

class MyApplication(asab.Application):

    async def initialize(self):
        self.add_module(asab.web.Module)  #(1)!
        self.WebService = self.get_service("asab.WebService") #(2)!

        self.WebContainer = asab.web.WebContainer( #(3)!
            websvc=self.WebService,
            config_section_name='my:web',
            config={"listen": "0.0.0.0:8080"}
        )

        self.WebContainer.WebApp.router.add_get('/hello', self.hello) #(4)!

    # This is the web request handler
    async def hello(self, request): #(5)!
        return aiohttp.web.Response(text='Hello from your ASAB server!\n')


if __name__ == '__main__':
    app = MyApplication()
    app.run()

1. In order to use asab.WebService, first import the corresponding asab.web.Module...
2. ...and then locate the Service.
3. Creates the Web container, which is instance of asab.Config.Configurable object that stores the configuration such as the actual web application.
4. asab.web.WebContainer.WebApp is instance of aiohttp.web.Application object. To create a new endpoint, use methods for aiohttp.web.Application.router object.
5. A request handler must be a coroutine that accepts aiohttp.web.Request instance as its only parameter and returns aiohttp.web.Response instance.

Tip

Resources may also have variable path, see the documentation.

web.get('/users/{user}/age', user_age_handler)
web.get(r'/{name:\d+}', name_handler)

Note

aiohttp also supports the Flask style for creating web request handlers via decorators. ASAB applications use the Django style way of creating routes, i.e. without decorators.

Web server configuration

Configuration is passed to the asab.web.WebContainer object.

Parameter	Meaning
listen	The socket address to which the web server will listen
backlog	A number of unaccepted connections that the system will allow before refusing new connections, see socket.socket.listen() for details
rootdir	The root path for the server. In case of many web containers, each one can implement a different root
servertokens	Controls whether 'Server' response header field is included ('full') or faked ('prod')
body_max_size	Client's maximum size in a request, in bytes. If a POST request exceeds this value, aiohttp.HTTPRequestEntityTooLarge exception is raised. See the documentation for more information
cors	Origin policy for Cross-Origin Resource Sharing. Empty disables CORS. * allows all origins. Otherwise a comma- and/or whitespace-separated allowlist. See the CORS section
cors_preflight_paths	Path prefixes (/foo/*) and exact paths that receive CORS headers, including OPTIONS preflight. Default /*. See the CORS section
cors_allow_headers	Allowed request headers advertised in CORS responses. See the CORS section
cors_allow_methods	Allowed HTTP methods advertised in CORS responses. See the CORS section
cors_allow_credentials	When yes, credentialed cross-origin requests are allowed. See the CORS section

The default configuration

[web]
listen=0.0.0.0 8080
backlog=128
rootdir=
servertokens=full
cors=
cors_preflight_paths=/*
cors_allow_headers=Authorization, Content-Type, X-App, X-Request-Id
cors_allow_methods=GET, POST, PUT, PATCH, DELETE, OPTIONS
cors_allow_credentials=no
body_max_size=1024**2

Socket addresses

The default configuration of the web socket address in asab.web.WebContainer is the following:

[web]
listen=0.0.0.0:8080

Multiple listening interfaces can be specified:

[web]
listen:
    0.0.0.0:8080
    :: 8080

Multiple listening interfaces, one with HTTPS (TLS/SSL) can be specified:

[web]
listen:
    0.0.0.0 8080
    :: 8080
    0.0.0.0 8443 ssl:web

[ssl:web]
cert=...
key=...
...

Multiple interfaces, one with HTTPS (inline):

[web]
listen:
    0.0.0.0 8080
    :: 8080
    0.0.0.0 8443 ssl

# The SSL parameters are inside of the WebContainer section
cert=...
key=...

You can also enable listening on TCP port 8080, IPv4 and IPv6 if applicable:

[web]
listen=8080

Reference

asab.web.create_web_server(app, section='web', config=None, api=False)

Build the web server with the specified configuration.

It is an user convenience function that simplifies typical way of how the web server is created.

Parameters:

Name	Type	Description	Default
app	Application	A reference to the ASAB Application.	required
section	str	Configuration section name with which the WebContainer will be created.	'web'
config	dict | None	Additional server configuration.	None

Returns:

Type	Description
UrlDispatcher	WebContainer Application Router object.

Examples:

class MyApplication(asab.Application):
        async def initialize(self):
                web = asab.web.create_web_server(self)
                web.add_get('/hello', self.hello)

        async def hello(self, request):
                return asab.web.rest.json_response(request, data="Hello, world!
")

Source code in asab/web/__init__.py

def create_web_server(app, section: str = "web", config: typing.Optional[dict] = None, api: bool = False) -> aiohttp.web.UrlDispatcher:
	"""
Build the web server with the specified configuration.

It is an user convenience function that simplifies typical way of how the web server is created.

Args:
	app (asab.Application): A reference to the ASAB Application.
	section (str): Configuration section name with which the WebContainer will be created.
	config (dict | None): Additional server configuration.

Returns:
	[WebContainer Application Router object](https://docs.aiohttp.org/en/stable/web_reference.html?highlight=router#router).

Examples:

```python
class MyApplication(asab.Application):
	async def initialize(self):
		web = asab.web.create_web_server(self)
		web.add_get('/hello', self.hello)

	async def hello(self, request):
		return asab.web.rest.json_response(request, data="Hello, world!\n")

```
	"""
	app.add_module(Module)
	websvc = app.get_service("asab.WebService")
	container = WebContainer(websvc, section, config=config)

	if api:
		# The DiscoverySession is functional only with ApiService initialized.
		from ..api import ApiService
		apisvc = ApiService(app)
		apisvc.initialize_web(container)

	return container.WebApp.router

asab.web.service.WebService

Bases: Service

Service for running and easy manipulation of the web server. It is used for registering and running the web container as well as initialization of web request metrics.

It should be used together with asab.web.WebContainer object that handles the web configuration.

Examples:

from asab.web import Module
self.add_module(Module)
web_service = self.get_service("asab.WebService")
container = asab.web.WebContainer(
        websvc=web_service,
        config_section_name='my:web',
        config={"listen": "0.0.0.0:8080"}
)

Source code in asab/web/service.py

class WebService(Service):
	"""
	Service for running and easy manipulation of the web server.
	It is used for registering and running the web container as well as initialization of web request metrics.

	It should be used together with [`asab.web.WebContainer`](#asab.web.WebContainer) object that handles the web configuration.

	Examples:

	```python
	from asab.web import Module
	self.add_module(Module)
	web_service = self.get_service("asab.WebService")
	container = asab.web.WebContainer(
		websvc=web_service,
		config_section_name='my:web',
		config={"listen": "0.0.0.0:8080"}
	)
	```
	"""

	ConfigSectionAliases = ["asab:web"]

	def __init__(self, app, service_name):
		super().__init__(app, service_name)

		# Web service is dependent on Metrics service
		if Config.getboolean("asab:metrics", "web_requests_metrics", fallback=False):
			app.add_module(metrics.Module)
			self.MetricsService = app.get_service("asab.MetricsService")
			self.WebRequestsMetrics = WebRequestsMetrics(self.MetricsService)
		self.Containers = {}

	async def finalize(self, app):
		for containers in self.Containers.values():
			await containers._stop(app)

	def _register_container(self, container, config_section_name: str):
		self.Containers[config_section_name] = container
		self.App.TaskService.schedule(container._start(self.App))


	@property
	def WebApp(self):
		"""
		An obsolete property only for maintaining backward compatibility. Please use `asab.web.WebContainer.WebApp` instead.
		"""
		return self.WebContainer.WebApp


	@property
	def WebContainer(self):
		"""
		An obsolete property only for maintaining backward compatibility. Please use `asab.web.WebContainer` instead.
		"""
		config_section = "web"

		# The WebContainer should be configured in the config section [web]
		if config_section not in Config.sections():
			# If there is no [web] section, try other aliases for backwards compatibility
			for alias in self.ConfigSectionAliases:
				if alias in Config.sections():
					config_section = alias
					L.warning(
						"Using obsolete web configuration section; migrate to [web].",
						struct_data={"section": alias},
					)
					break
			else:
				raise RuntimeError("No [web] section configured.")

		try:
			return self.Containers[config_section]
		except KeyError:
			from .container import WebContainer
			return WebContainer(self, config_section)

WebApp property

An obsolete property only for maintaining backward compatibility. Please use asab.web.WebContainer.WebApp instead.

WebContainer property

An obsolete property only for maintaining backward compatibility. Please use asab.web.WebContainer instead.

asab.web.WebContainer

Bases: Configurable

Configurable object that serves as a backend for asab.WebService. It contains everything needed for the web server existence, namely all the configuration and the server Application object.

Source code in asab/web/container.py

class WebContainer(Configurable):
	"""
	Configurable object that serves as a backend for `asab.WebService`.
	It contains everything needed for the web server existence, namely all the configuration and the server Application object.
	"""

	ConfigDefaults = {
		'listen': '0.0.0.0 8080',  # Can be multiline
		'backlog': 128,
		'rootdir': '',
		'servertokens': 'full',  # Controls whether 'Server' response header field is included ('full') or faked 'prod' ()
		'cors': '',
		'cors_preflight_paths': '/*',
		'cors_allow_headers': 'Authorization, Content-Type, X-App, X-Request-Id',
		'cors_allow_methods': 'GET, POST, PUT, PATCH, DELETE, OPTIONS',
		'cors_allow_credentials': 'no',
		'body_max_size': 1024**2,  # Client’s maximum body size in a request, in bytes
	}


	def __init__(self, websvc: WebService, config_section_name: str, config: typing.Optional[dict] = None):
		super().__init__(config_section_name=config_section_name, config=config)

		self.Addresses = None  # The address is available only after `WebContainer.started!` PubSub message is published.
		self.BackLog = int(self.Config.get("backlog"))
		self.CORS = cors.normalize_cors_config(self.Config.get("cors"))
		self.CORSHandler = None
		self._CORSPreflightRoutes = set()

		servertokens = self.Config.get("servertokens")
		if servertokens == 'prod':
			# Because we cannot remove token completely
			self.ServerTokens = "asab"
		else:
			from .. import __version__
			self.ServerTokens = aiohttp.web_response.SERVER_SOFTWARE + " asab/" + __version__

		# Parse listen address(es), can be multiline configuration item
		ls = self.Config.get("listen")

		if isinstance(ls, int):
			ls = str(ls)  # Assume that the integer is a port number
		if not isinstance(ls, str):
			raise TypeError("Invalid type of 'listen' configuration item: {}".format(type(ls)))

		self._listen = []
		for line in ls.split('\n'):
			line = line.strip()
			if len(line) == 0:
				continue

			if ' ' in line:
				line = re.split(r"\s+", line)
			else:
				# This line allows the (obsolete) format of IPv4 with ':'
				# such as "0.0.0.0:8001"
				line = re.split(r"[:\s]", line, 1)

			if all([c in '0123456789' for c in line[0]]):
				# If the first item is a number, consider that a port number
				addr = ["0.0.0.0", "::"]  # We want to listen on IPv4 and IPv6
				port = line.pop(0).strip()
				port = int(port)
			else:
				# First item is a port, a second is an IP address of the network interface to listen to
				addr = line.pop(0).strip()
				port = line.pop(0).strip()
				port = int(port)
			ssl_context = None

			for param in line:
				if param.startswith('ssl:'):
					# Dedicated section for SSL
					ssl_context = SSLContextBuilder(param, config=self.Config).build()
					# SSL parameters are included in the current config section
				elif param.startswith('ssl'):
					ssl_context = SSLContextBuilder("<none>", config=self.Config).build()
				else:
					raise RuntimeError(
						"Unknown listen parameter in section [{}]: {}".format(config_section_name, param)
					)

			if isinstance(addr, list):
				for a in addr:
					self._listen.append((a, port, ssl_context))
			else:
				self._listen.append((addr, port, ssl_context))

		if len(self._listen) == 0:
			L.warning(
				"Web container has no listen address configured; HTTP server will not start.",
				struct_data={"config_section": config_section_name},
			)

		client_max_size = int(self.Config.get("body_max_size"))
		self.WebApp: aiohttp.web.Application = aiohttp.web.Application(client_max_size=client_max_size)
		"""
		The Web Application object. See [aiohttp documentation](https://docs.aiohttp.org/en/stable/web_reference.html?highlight=Application#application) for the details.

		It is a *dict-like* object, so you can use it for sharing data globally by storing arbitrary properties for later access from a handler.

		Attributes:
			WebApp["app"] (asab.Application): Reference to the ASAB Application.

			WebApp["rootdir"] (asab.web.staticdir.StaticDirProvider): Reference to the root path specified by `rootdir` configuration.
		"""
		self.WebApp.on_response_prepare.append(self._on_prepare_response)
		self.WebApp['app'] = websvc.App

		rootdir = self.Config.get("rootdir")
		if len(rootdir) > 0:
			from .staticdir import StaticDirProvider
			self.WebApp['rootdir'] = StaticDirProvider(self.WebApp, root='/', path=rootdir)

		access_log = logging.getLogger(__name__[:__name__.rfind('.')] + '.al')
		access_log.App = websvc.App

		self.WebAppRunner = aiohttp.web.AppRunner(
			self.WebApp,
			handle_signals=False,
			access_log=access_log,
			access_log_class=AccessLogger,
		)

		websvc._register_container(self, config_section_name)

		if self.CORS:
			self.enable_cors(allow_origin=self.CORS)

		self.WebApp.middlewares.append(set_request_context)


	async def _start(self, app: Application):
		await self.WebAppRunner.setup()

		for addr, port, ssl_context in self._listen:
			site = aiohttp.web.TCPSite(
				self.WebAppRunner,
				host=addr, port=port, backlog=self.BackLog,
				ssl_context=ssl_context,
			)
			try:
				await site.start()
			except OSError as err:
				L.error(
					"Web server failed to bind to the configured listen address.",
					struct_data={"address": addr, "port": port, "error": str(err)},
				)

			if isinstance(site, aiohttp.web_runner.TCPSite):
				for address in site._runner.addresses:
					if self.Addresses is None:
						self.Addresses = []
					self.Addresses.append(address)

		self.WebApp['app'].PubSub.publish("WebContainer.started!", self)


	async def _stop(self, app: Application):
		self.WebApp['app'].PubSub.publish("WebContainer.stopped!", self)
		await self.WebAppRunner.cleanup()


	def enable_cors(
		self,
		allow_origin: typing.Union[str, typing.Iterable[str], typing.Callable[[str], bool]],
		preflight_paths: typing.Union[str, typing.Iterable[str], None] = None,
		allow_headers: typing.Union[str, typing.Iterable[str], None] = None,
		allow_methods: typing.Union[str, typing.Iterable[str], None] = None,
		allow_credentials: typing.Optional[bool] = None,
	):
		"""
		Enable Cross-Origin Resource Sharing on this web container.

		If `[web] cors` is non-empty, this method is called automatically during container
		construction. Applications such as SeaCat Auth can call it from code instead
		(leave `cors` empty and pass a callable origin check).

		Calling this method again replaces the origin policy and adds any new preflight
		paths. OPTIONS routes that are already registered are not added twice.

		Args:
			allow_origin: `"*"` to allow every origin, a string or iterable of allowed
				origins, or a callable `origin: str -> bool` for a dynamic allowlist.
			preflight_paths: Path prefixes (`/foo/*`) and exact paths that receive CORS
				headers, including OPTIONS preflight. Defaults to `[web] cors_preflight_paths`.
			allow_headers: Allowed request headers. Defaults to `[web] cors_allow_headers`.
			allow_methods: Allowed HTTP methods. Defaults to `[web] cors_allow_methods`.
			allow_credentials: Whether browsers may send cookies and Authorization.
				Defaults to `[web] cors_allow_credentials`. When this is true, the
				response echoes the request `Origin` even if `allow_origin` is `"*"`;
				`Access-Control-Allow-Origin: *` is never combined with credentials.
		"""
		if preflight_paths is None:
			preflight_paths = self.Config.get("cors_preflight_paths")
		if allow_headers is None:
			allow_headers = self.Config.get("cors_allow_headers")
		if allow_methods is None:
			allow_methods = self.Config.get("cors_allow_methods")
		if allow_credentials is None:
			allow_credentials = self.Config.getboolean("cors_allow_credentials")

		# Normalize paths before touching any state so a bad value cannot leave
		# the handler half-updated.
		preflight_paths = cors.normalize_path_list(preflight_paths)

		if self.CORSHandler is None:
			self.CORSHandler = cors.CORSHandler(
				allow_origin=allow_origin,
				paths=preflight_paths,
				allow_headers=allow_headers,
				allow_methods=allow_methods,
				allow_credentials=allow_credentials,
			)
		else:
			self.CORSHandler.set_policy(
				allow_origin,
				allow_headers,
				allow_methods,
				allow_credentials,
			)
			self.CORSHandler.add_paths(preflight_paths)

		self._register_preflight_routes()


	def add_preflight_handlers(self, preflight_paths: typing.Iterable[str]):
		"""
		Add OPTIONS handlers and CORS path patterns to already-enabled CORS.

		Use `enable_cors()` to start CORS. This method only extends the set of paths.

		Args:
			preflight_paths: Path prefixes (`/foo/*`) and exact paths that should
				receive CORS, including OPTIONS preflight.
		"""
		if self.CORSHandler is None:
			raise RuntimeError("CORS is not enabled; call enable_cors() first.")
		self.CORSHandler.add_paths(preflight_paths)
		self._register_preflight_routes()


	def _register_preflight_routes(self):
		"""
		Register OPTIONS routes for all CORS path patterns.

		The handler's `Paths` list is the single source of truth for path patterns;
		this method reads it, so it is idempotent and never registers the same route
		twice. Paths ending in `/*` produce two routes: the glob (`/foo/{tail:.*}`)
		and the directory itself (`/foo`), because aiohttp's `{tail:.*}` does not
		match the bare directory. Root `/*` is exempt because `/{tail:.*}` already
		covers `/`.
		"""
		for path in self.CORSHandler.Paths:
			route_paths = [cors.path_to_route(path)]
			if path.endswith("/*") and path != "/*":
				directory = path[:-2]
				if directory:
					route_paths.append(directory)
			for route_path in route_paths:
				if route_path in self._CORSPreflightRoutes:
					continue
				self.WebApp.router.add_route("OPTIONS", route_path, self._preflight_handler)
				self._CORSPreflightRoutes.add(route_path)


	async def _preflight_handler(self, request):
		"""
		Answer CORS preflight requests with 204 No Content.

		The CORS headers are applied in `_on_prepare_response`, so preflight and
		actual responses share the same policy. A preflight request still receives
		204 even when the `Origin` header is missing or the origin is not allowed;
		in that case no CORS headers are attached.
		"""
		return aiohttp.web.HTTPNoContent()


	async def _on_prepare_response(self, request, response):
		"""
		Attach server and CORS headers to every outgoing response.

		CORS headers are attached only when CORS is enabled and the request origin
		and path match the policy; see `CORSHandler.apply()`.
		"""
		response.headers['Server'] = self.ServerTokens

		if self.CORSHandler is not None:
			self.CORSHandler.apply(request, response)


	def get_ports(self) -> typing.List[str]:
		"""
		Return list of available ports.

		Returns:
			(list[str]) List of ports.
		"""
		ports = []
		for addr, port, ssl_context in self._listen:
			ports.append(port)
		return ports

WebApp = aiohttp.web.Application(client_max_size=client_max_size) instance-attribute

The Web Application object. See aiohttp documentation for the details.

It is a dict-like object, so you can use it for sharing data globally by storing arbitrary properties for later access from a handler.

Attributes:

Name	Type	Description
WebApp["app"]	Application	Reference to the ASAB Application.
WebApp["rootdir"]	StaticDirProvider	Reference to the root path specified by rootdir configuration.

add_preflight_handlers(preflight_paths)

Add OPTIONS handlers and CORS path patterns to already-enabled CORS.

Use enable_cors() to start CORS. This method only extends the set of paths.

Parameters:

Name	Type	Description	Default
preflight_paths	Iterable[str]	Path prefixes (/foo/*) and exact paths that should receive CORS, including OPTIONS preflight.	required

Source code in asab/web/container.py

def add_preflight_handlers(self, preflight_paths: typing.Iterable[str]):
	"""
	Add OPTIONS handlers and CORS path patterns to already-enabled CORS.

	Use `enable_cors()` to start CORS. This method only extends the set of paths.

	Args:
		preflight_paths: Path prefixes (`/foo/*`) and exact paths that should
			receive CORS, including OPTIONS preflight.
	"""
	if self.CORSHandler is None:
		raise RuntimeError("CORS is not enabled; call enable_cors() first.")
	self.CORSHandler.add_paths(preflight_paths)
	self._register_preflight_routes()

enable_cors(allow_origin, preflight_paths=None, allow_headers=None, allow_methods=None, allow_credentials=None)

Enable Cross-Origin Resource Sharing on this web container.

If [web] cors is non-empty, this method is called automatically during container construction. Applications such as SeaCat Auth can call it from code instead (leave cors empty and pass a callable origin check).

Calling this method again replaces the origin policy and adds any new preflight paths. OPTIONS routes that are already registered are not added twice.

Parameters:

Name	Type	Description	Default
allow_origin	Union[str, Iterable[str], Callable[[str], bool]]	"*" to allow every origin, a string or iterable of allowed origins, or a callable origin: str -> bool for a dynamic allowlist.	required
preflight_paths	Union[str, Iterable[str], None]	Path prefixes (/foo/*) and exact paths that receive CORS headers, including OPTIONS preflight. Defaults to [web] cors_preflight_paths.	None
allow_headers	Union[str, Iterable[str], None]	Allowed request headers. Defaults to [web] cors_allow_headers.	None
allow_methods	Union[str, Iterable[str], None]	Allowed HTTP methods. Defaults to [web] cors_allow_methods.	None
allow_credentials	Optional[bool]	Whether browsers may send cookies and Authorization. Defaults to [web] cors_allow_credentials. When this is true, the response echoes the request Origin even if allow_origin is "*"; Access-Control-Allow-Origin: * is never combined with credentials.	None

Source code in asab/web/container.py

def enable_cors(
	self,
	allow_origin: typing.Union[str, typing.Iterable[str], typing.Callable[[str], bool]],
	preflight_paths: typing.Union[str, typing.Iterable[str], None] = None,
	allow_headers: typing.Union[str, typing.Iterable[str], None] = None,
	allow_methods: typing.Union[str, typing.Iterable[str], None] = None,
	allow_credentials: typing.Optional[bool] = None,
):
	"""
	Enable Cross-Origin Resource Sharing on this web container.

	If `[web] cors` is non-empty, this method is called automatically during container
	construction. Applications such as SeaCat Auth can call it from code instead
	(leave `cors` empty and pass a callable origin check).

	Calling this method again replaces the origin policy and adds any new preflight
	paths. OPTIONS routes that are already registered are not added twice.

	Args:
		allow_origin: `"*"` to allow every origin, a string or iterable of allowed
			origins, or a callable `origin: str -> bool` for a dynamic allowlist.
		preflight_paths: Path prefixes (`/foo/*`) and exact paths that receive CORS
			headers, including OPTIONS preflight. Defaults to `[web] cors_preflight_paths`.
		allow_headers: Allowed request headers. Defaults to `[web] cors_allow_headers`.
		allow_methods: Allowed HTTP methods. Defaults to `[web] cors_allow_methods`.
		allow_credentials: Whether browsers may send cookies and Authorization.
			Defaults to `[web] cors_allow_credentials`. When this is true, the
			response echoes the request `Origin` even if `allow_origin` is `"*"`;
			`Access-Control-Allow-Origin: *` is never combined with credentials.
	"""
	if preflight_paths is None:
		preflight_paths = self.Config.get("cors_preflight_paths")
	if allow_headers is None:
		allow_headers = self.Config.get("cors_allow_headers")
	if allow_methods is None:
		allow_methods = self.Config.get("cors_allow_methods")
	if allow_credentials is None:
		allow_credentials = self.Config.getboolean("cors_allow_credentials")

	# Normalize paths before touching any state so a bad value cannot leave
	# the handler half-updated.
	preflight_paths = cors.normalize_path_list(preflight_paths)

	if self.CORSHandler is None:
		self.CORSHandler = cors.CORSHandler(
			allow_origin=allow_origin,
			paths=preflight_paths,
			allow_headers=allow_headers,
			allow_methods=allow_methods,
			allow_credentials=allow_credentials,
		)
	else:
		self.CORSHandler.set_policy(
			allow_origin,
			allow_headers,
			allow_methods,
			allow_credentials,
		)
		self.CORSHandler.add_paths(preflight_paths)

	self._register_preflight_routes()

get_ports()

Return list of available ports.

Returns:

Type	Description
List[str]	(list[str]) List of ports.

Source code in asab/web/container.py

def get_ports(self) -> typing.List[str]:
	"""
	Return list of available ports.

	Returns:
		(list[str]) List of ports.
	"""
	ports = []
	for addr, port, ssl_context in self._listen:
		ports.append(port)
	return ports