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/channels2/router.py

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.channels2.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)

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


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."
            )

    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
        )

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

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

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