Skip to content

base

Base classes for database clients.

AllDbBase

Base class (not an ABC, but could be) for all Db clients.

Source code in src/embar/db/base.py 
12
13
14
15
16
17
class AllDbBase:
    """
    Base class (not an ABC, but could be) for all Db clients.
    """

    db_type: DbType = Undefined

AsyncDbBase

Bases: ABC, AllDbBase

Base class for async Db clients.

Source code in src/embar/db/base.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
class AsyncDbBase(ABC, AllDbBase):
    """
    Base class for async Db clients.
    """

    @abstractmethod
    async def execute(self, query: QuerySingle):
        """
        Execute a query without returning results.
        """
        ...

    @abstractmethod
    async def executemany(self, query: QueryMany):
        """
        Execute a query with multiple parameter sets.
        """
        ...

    @abstractmethod
    async def fetch(self, query: QuerySingle | QueryMany) -> list[dict[str, Any]]:
        """
        Execute a query and return results as a list of dicts.
        """
        ...

    @abstractmethod
    async def truncate(self, schema: str | None = None) -> None:
        """
        Truncate all tables in the schema.
        """
        ...

    @abstractmethod
    async def drop_tables(self, schema: str | None = None) -> None:
        """
        Drop all tables in the schema.
        """
        ...

drop_tables(schema=None) abstractmethod async

Drop all tables in the schema.

Source code in src/embar/db/base.py 
94
95
96
97
98
99
@abstractmethod
async def drop_tables(self, schema: str | None = None) -> None:
    """
    Drop all tables in the schema.
    """
    ...

execute(query) abstractmethod async

Execute a query without returning results.

Source code in src/embar/db/base.py 
66
67
68
69
70
71
@abstractmethod
async def execute(self, query: QuerySingle):
    """
    Execute a query without returning results.
    """
    ...

executemany(query) abstractmethod async

Execute a query with multiple parameter sets.

Source code in src/embar/db/base.py 
73
74
75
76
77
78
@abstractmethod
async def executemany(self, query: QueryMany):
    """
    Execute a query with multiple parameter sets.
    """
    ...

fetch(query) abstractmethod async

Execute a query and return results as a list of dicts.

Source code in src/embar/db/base.py 
80
81
82
83
84
85
@abstractmethod
async def fetch(self, query: QuerySingle | QueryMany) -> list[dict[str, Any]]:
    """
    Execute a query and return results as a list of dicts.
    """
    ...

truncate(schema=None) abstractmethod async

Truncate all tables in the schema.

Source code in src/embar/db/base.py 
87
88
89
90
91
92
@abstractmethod
async def truncate(self, schema: str | None = None) -> None:
    """
    Truncate all tables in the schema.
    """
    ...

DbBase

Bases: ABC, AllDbBase

Base class for sync Db clients.

Source code in src/embar/db/base.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
class DbBase(ABC, AllDbBase):
    """
    Base class for _sync_ Db clients.
    """

    @abstractmethod
    def execute(self, query: QuerySingle):
        """
        Execute a query without returning results.
        """
        ...

    @abstractmethod
    def executemany(self, query: QueryMany):
        """
        Execute a query with multiple parameter sets.
        """
        ...

    @abstractmethod
    def fetch(self, query: QuerySingle | QueryMany) -> list[dict[str, Any]]:
        """
        Execute a query and return results as a list of dicts.
        """
        ...

    @abstractmethod
    def truncate(self, schema: str | None = None) -> None:
        """
        Truncate all tables in the schema.
        """
        ...

    @abstractmethod
    def drop_tables(self, schema: str | None = None) -> None:
        """
        Drop all tables in the schema.
        """
        ...

drop_tables(schema=None) abstractmethod

Drop all tables in the schema.

Source code in src/embar/db/base.py 
53
54
55
56
57
58
@abstractmethod
def drop_tables(self, schema: str | None = None) -> None:
    """
    Drop all tables in the schema.
    """
    ...

execute(query) abstractmethod

Execute a query without returning results.

Source code in src/embar/db/base.py 
25
26
27
28
29
30
@abstractmethod
def execute(self, query: QuerySingle):
    """
    Execute a query without returning results.
    """
    ...

executemany(query) abstractmethod

Execute a query with multiple parameter sets.

Source code in src/embar/db/base.py 
32
33
34
35
36
37
@abstractmethod
def executemany(self, query: QueryMany):
    """
    Execute a query with multiple parameter sets.
    """
    ...

fetch(query) abstractmethod

Execute a query and return results as a list of dicts.

Source code in src/embar/db/base.py 
39
40
41
42
43
44
@abstractmethod
def fetch(self, query: QuerySingle | QueryMany) -> list[dict[str, Any]]:
    """
    Execute a query and return results as a list of dicts.
    """
    ...

truncate(schema=None) abstractmethod

Truncate all tables in the schema.

Source code in src/embar/db/base.py 
46
47
48
49
50
51
@abstractmethod
def truncate(self, schema: str | None = None) -> None:
    """
    Truncate all tables in the schema.
    """
    ...