• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

lorinkoz / django-pgschemas / 6864446688

14 Nov 2023 01:40PM UTC coverage: 15.956% (-77.7%) from 93.617%
6864446688

push

github

lorinkoz
Baby steps in a huge refactoring

51 of 76 new or added lines in 4 files covered. (67.11%)

744 existing lines in 20 files now uncovered.

232 of 1454 relevant lines covered (15.96%)

1.91 hits per line

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

65.22
/django_pgschemas/schema.py
1
from contextlib import contextmanager
12✔
2
from contextvars import ContextVar
12✔
3
from typing import Any, Iterator
12✔
4

5
from asgiref.sync import sync_to_async
12✔
6

7
from django_pgschemas.signals import schema_activate
12✔
8

9

10
class Schema:
12✔
11
    schema_name = None
12✔
12
    domain_url = None
12✔
13
    folder = None
12✔
14

15
    is_dynamic = False
12✔
16

17
    @staticmethod
12✔
18
    def create(schema_name: str, domain_url: str | None = None, folder: str | None = None):
12✔
19
        schema = Schema()
12✔
20
        schema.schema_name = schema_name
12✔
21
        schema.domain_url = domain_url
12✔
22
        schema.folder = folder
12✔
23
        return schema
12✔
24

25
    def __enter__(self) -> None:
12✔
UNCOV
26
        schema = active.get()
×
UNCOV
27
        if schema is not None:
×
UNCOV
28
            self._previous_active_token = active.set(self)
×
29

30
    __aenter__ = sync_to_async(__enter__)
12✔
31

32
    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
12✔
UNCOV
33
        _previous_active_token = getattr(self, "_previous_active_token", None)
×
UNCOV
34
        if _previous_active_token is not None:
×
UNCOV
35
            active.reset(_previous_active_token)
×
36

37
    __aexit__ = sync_to_async(__exit__)
12✔
38

39

40
def get_default_schema() -> "Schema":
12✔
41
    return Schema.create("public")
12✔
42

43

44
active: ContextVar["Schema"] = ContextVar("active", default=get_default_schema())
12✔
45

46

47
def get_current_schema() -> "Schema":
12✔
UNCOV
48
    return active.get()
×
49

50

51
def activate(schema: "Schema") -> None:
12✔
UNCOV
52
    if not isinstance(schema, Schema):
×
UNCOV
53
        raise RuntimeError("'activate' must be called with a Schema descendant")
×
54

UNCOV
55
    active.set(schema)
×
56

UNCOV
57
    schema_activate.send(sender=Schema, schema=schema)
×
58

59

60
def deactivate() -> None:
12✔
UNCOV
61
    active.set(get_default_schema())
×
62

UNCOV
63
    schema_activate.send(sender=Schema, schema=Schema.create("public"))
×
64

65

66
activate_public = deactivate
12✔
67

68

69
@contextmanager
12✔
70
def override(schema: "Schema") -> Iterator[None]:
12✔
UNCOV
71
    token = active.set(schema)
×
72

UNCOV
73
    yield
×
74

UNCOV
75
    active.reset(token)
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2024 Coveralls, Inc