10262409303

Committed 06 Aug 2024 07:45AM UTC coverage: 58.477%. Remained the same

Build # 10262409303

Build Type

push

github

Committed by

lorinkoz

Commit Message

Bump mypy from 1.11.0 to 1.11.1

Bumps [mypy](https://github.com/python/mypy) from 1.11.0 to 1.11.1.
- [Changelog](https://github.com/python/mypy/blob/master/CHANGELOG.md)
- [Commits](https://github.com/python/mypy/compare/v1.11...v1.11.1)

---
updated-dependencies:
- dependency-name: mypy
  dependency-type: direct:development
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>

Run Details

914 of 1563 relevant lines covered (58.48%)

7.01 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/django_pgschemas/contrib/channels3/router.py

from channels.db import database_sync_to_async
from channels.routing import ProtocolTypeRouter, URLRouter
from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.utils.encoding import force_str
from django.utils.module_loading import import_string

from django_pgschemas.contrib.channels3.auth import TenantAuthMiddlewareStack
from django_pgschemas.routing.info import DomainInfo
from django_pgschemas.schema import Schema
from django_pgschemas.utils import get_domain_model, remove_www


class TenantAwareProtocolTypeRouter(ProtocolTypeRouter):
    def __init__(self, application_mapping, tenant_prefix):
        self.tenant_prefix = tenant_prefix
        super().__init__(application_mapping)

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            scope["path"] = scope["path"][len(self.tenant_prefix) + 1 :]
        return await super().__call__(scope, receive, send)


class TenantProtocolRouter:
    """
    ProtocolRouter that handles multi-tenancy.
    """

    def __init__(self):
        self.root_ws_urlconf = settings.TENANTS["default"].get("WS_URLCONF")
        if self.root_ws_urlconf is None:
            raise ImproperlyConfigured(
                "TENANTS['default'] must contain a 'WS_URLCONF' key in order to use TenantProtocolRouter."
            )

    @database_sync_to_async
    def get_tenant_scope(self, scope):
        """
        Get tenant and websockets urlconf based on scope host.
        """
        hostname = force_str(dict(scope["headers"]).get(b"host", b""))
        hostname = remove_www(hostname.split(":")[0])

        tenant = None
        ws_urlconf = self.root_ws_urlconf

        # Checking for static tenants
        for schema, data in settings.TENANTS.items():
            if schema in ["public", "default"]:
                continue
            if hostname in data.get("DOMAINS", []):
                tenant = Schema.create(
                    schema_name=schema,
                    routing=DomainInfo(domain=hostname),
                )
                if "WS_URLCONF" in data:
                    ws_urlconf = data["WS_URLCONF"]
                return tenant, "", import_string(ws_urlconf + ".urlpatterns")

        # Checking for dynamic tenants
        else:
            DomainModel = get_domain_model()
            prefix = scope["path"].split("/")[1]
            try:
                domain = DomainModel.objects.select_related("tenant").get(
                    domain=hostname, folder=prefix
                )
            except DomainModel.DoesNotExist:
                try:
                    domain = DomainModel.objects.select_related("tenant").get(
                        domain=hostname, folder=""
                    )
                except DomainModel.DoesNotExist:
                    return None, "", []
            tenant = domain.tenant
            tenant.domain_url = hostname
            ws_urlconf = settings.TENANTS["default"]["WS_URLCONF"]
            return (
                tenant,
                prefix if prefix == domain.folder else "",
                import_string(ws_urlconf + ".urlpatterns"),
            )

    def get_protocol_type_router(self, tenant_prefix, ws_urlconf):
        """
        Subclasses can override this to include more protocols.
        """
        return TenantAwareProtocolTypeRouter(
            {"websocket": TenantAuthMiddlewareStack(URLRouter(ws_urlconf))}, tenant_prefix
        )

    async def __call__(self, scope, receive, send):
        tenant, tenant_prefix, ws_urlconf = await self.get_tenant_scope(scope)
        scope.update({"tenant": tenant})
        return await self.get_protocol_type_router(tenant_prefix, ws_urlconf)(scope, receive, send)

1	from channels.db import database_sync_to_async	×
2	from channels.routing import ProtocolTypeRouter, URLRouter	×
3	from django.conf import settings	×
4	from django.core.exceptions import ImproperlyConfigured	×
5	from django.utils.encoding import force_str	×
6	from django.utils.module_loading import import_string	×
7
8	from django_pgschemas.contrib.channels3.auth import TenantAuthMiddlewareStack	×
9	from django_pgschemas.routing.info import DomainInfo	×
10	from django_pgschemas.schema import Schema	×
11	from django_pgschemas.utils import get_domain_model, remove_www	×
12
13
14	class TenantAwareProtocolTypeRouter(ProtocolTypeRouter):	×
15	def __init__(self, application_mapping, tenant_prefix):	×
16	self.tenant_prefix = tenant_prefix	×
17	super().__init__(application_mapping)	×
18
19	async def __call__(self, scope, receive, send):	×
20	if scope["type"] != "http":	×
21	scope["path"] = scope["path"][len(self.tenant_prefix) + 1 :]	×
22	return await super().__call__(scope, receive, send)	×
23
24
25	class TenantProtocolRouter:	×
26	"""
27	ProtocolRouter that handles multi-tenancy.
28	"""
29
30	def __init__(self):	×
31	self.root_ws_urlconf = settings.TENANTS["default"].get("WS_URLCONF")	×
32	if self.root_ws_urlconf is None:	×
33	raise ImproperlyConfigured(	×
34	"TENANTS['default'] must contain a 'WS_URLCONF' key in order to use TenantProtocolRouter."
35	)
36
37	@database_sync_to_async	×
38	def get_tenant_scope(self, scope):	×
39	"""
40	Get tenant and websockets urlconf based on scope host.
41	"""
42	hostname = force_str(dict(scope["headers"]).get(b"host", b""))	×
43	hostname = remove_www(hostname.split(":")[0])	×
44
45	tenant = None	×
46	ws_urlconf = self.root_ws_urlconf	×
47
48	# Checking for static tenants
49	for schema, data in settings.TENANTS.items():	×
50	if schema in ["public", "default"]:	×
51	continue	×
52	if hostname in data.get("DOMAINS", []):	×
53	tenant = Schema.create(	×
54	schema_name=schema,
55	routing=DomainInfo(domain=hostname),
56	)
57	if "WS_URLCONF" in data:	×
58	ws_urlconf = data["WS_URLCONF"]	×
59	return tenant, "", import_string(ws_urlconf + ".urlpatterns")	×
60
61	# Checking for dynamic tenants
62	else:
63	DomainModel = get_domain_model()	×
64	prefix = scope["path"].split("/")[1]	×
65	try:	×
66	domain = DomainModel.objects.select_related("tenant").get(	×
67	domain=hostname, folder=prefix
68	)
69	except DomainModel.DoesNotExist:	×
70	try:	×
71	domain = DomainModel.objects.select_related("tenant").get(	×
72	domain=hostname, folder=""
73	)
74	except DomainModel.DoesNotExist:	×
75	return None, "", []	×
76	tenant = domain.tenant	×
77	tenant.domain_url = hostname	×
78	ws_urlconf = settings.TENANTS["default"]["WS_URLCONF"]	×
79	return (	×
80	tenant,
81	prefix if prefix == domain.folder else "",
82	import_string(ws_urlconf + ".urlpatterns"),
83	)
84
85	def get_protocol_type_router(self, tenant_prefix, ws_urlconf):	×
86	"""
87	Subclasses can override this to include more protocols.
88	"""
89	return TenantAwareProtocolTypeRouter(	×
90	{"websocket": TenantAuthMiddlewareStack(URLRouter(ws_urlconf))}, tenant_prefix
91	)
92
93	async def __call__(self, scope, receive, send):	×
94	tenant, tenant_prefix, ws_urlconf = await self.get_tenant_scope(scope)	×
95	scope.update({"tenant": tenant})	×
96	return await self.get_protocol_type_router(tenant_prefix, ws_urlconf)(scope, receive, send)	×

lorinkoz / django-pgschemas / 10262409303

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous