Skip to content

API Reference

AllDatabaseSettings

Bases: BaseModel

Settings for the database connection.

Source code in db4me/settings.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class AllDatabaseSettings(BaseModel):
    """
    Settings for the database connection.
    """

    db: DatabaseSettings = Field(
        default_factory=cast(Callable[[], Any], DatabaseSettings),
        description="Database settings.",
    )
    pg_db: PgDatabaseSettings = Field(
        default_factory=PgDatabaseSettings,
        description="Database settings specific for postgres.",
    )
    sqlite_db: SqliteDatabaseSettings = Field(
        default_factory=SqliteDatabaseSettings,
        description="Database settings specific for SqLite.",
    )

ConfigurationError

Bases: Db4MeError

Raised when there is an error in the configuration file.

Source code in db4me/errors.py 
 7
 8
 9
10
class ConfigurationError(Db4MeError):
    """Raised when there is an error in the configuration file."""

    pass

DatabaseSettings

Bases: BaseModel

Common settings for all database types.

Attributes:

Name Type Description
url Optional[str]

The database URL. In case of async connections this is where the async database URL should be set.
sync_alternative Optional[str]

The database URL to use for sync connections. This should be used when an async connection exists in url. If missing an attempt will be made to convert the async URL to a sync URL.
echo Optional[bool]

If True, the engine will log all statements as well as a repr() of their parameter lists to the engines logger, which defaults to sys.stdout.
pool_size Optional[int]

The size of the database pool.
pool_recycle Optional[int]

The number of seconds after which a database connection should be recycled.
pool_pre_ping Optional[bool]

If True, the database connection pool will be pre-pinged This means that the connection will be checked if it is still alive before it is used.
pool_reset_on_return Optional[str]

The action to take when a connection is returned to the pool. Possible values are: - "rollback": Rollback the transaction. - "commit": Commit the transaction. - "disconnect": Disconnect the connection.
Source code in db4me/settings.py 
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
59
60
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
class DatabaseSettings(BaseModel):
    """
    Common settings for all database types.

    Attributes:
        url: The database URL. In case of async connections this is where the
            async database URL should be set.
        sync_alternative: The database URL to use for sync connections. This
            should be used when an async connection exists in `url`. If missing
            an attempt will be made to convert the async URL to a sync URL.
        echo: If True, the engine will log all statements as well as a
            repr() of their parameter lists to the engines logger, which
            defaults to sys.stdout.
        pool_size: The size of the database pool.
        pool_recycle: The number of seconds after which a database connection
            should be recycled.
        pool_pre_ping: If True, the database connection pool will be pre-pinged
            This means that the connection will be checked if it is still
            alive before it is used.
        pool_reset_on_return: The action to take when a connection is returned
            to the pool. Possible values are:
                - "rollback": Rollback the transaction.
                - "commit": Commit the transaction.
                - "disconnect": Disconnect the connection.
    """

    # Database URL
    url: Optional[str] = "sqlite://"

    # Database alternative URL.
    sync_alternative: Optional[str] = Field(None, validate_default=True)

    # Database echo
    echo: Optional[bool] = False

    # Database pool size
    pool_size: Optional[int] = 10

    # Database pool recycle
    pool_recycle: Optional[int] = 3600

    # Database pool pre ping
    pool_pre_ping: Optional[bool] = True

    # Database pool reset on return
    pool_reset_on_return: Optional[str] = "rollback"

    @field_validator("sync_alternative")
    @classmethod
    def validate_sync_alternative(cls, v, values: ValidationInfo):
        """
        Validate the sync_alternative field.
        """
        url = values.data.get("url")
        if v is None and url is not None:
            if url.startswith("postgresql+asyncpg"):
                v = url.replace("postgresql+asyncpg", "postgresql")
            elif url.startswith("sqlite+aiosqlite"):
                v = url.replace("sqlite+aiosqlite", "sqlite")
        return v

validate_sync_alternative(v, values) classmethod

Validate the sync_alternative field.

Source code in db4me/settings.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
@field_validator("sync_alternative")
@classmethod
def validate_sync_alternative(cls, v, values: ValidationInfo):
    """
    Validate the sync_alternative field.
    """
    url = values.data.get("url")
    if v is None and url is not None:
        if url.startswith("postgresql+asyncpg"):
            v = url.replace("postgresql+asyncpg", "postgresql")
        elif url.startswith("sqlite+aiosqlite"):
            v = url.replace("sqlite+aiosqlite", "sqlite")
    return v

Db4MeError

Bases: Exception

Base class for exceptions in this module.

Source code in db4me/errors.py 
1
2
3
4
class Db4MeError(Exception):
    """Base class for exceptions in this module."""

    pass

PgDatabaseSettings

Bases: BaseModel

Settings for connections to postgres databases.

Source code in db4me/settings.py 
18
19
20
21
22
23
class PgDatabaseSettings(BaseModel):
    """
    Settings for connections to postgres databases.
    """

    pass

SqliteDatabaseSettings

Bases: BaseModel

Settings for connections to SqLite databases.

Source code in db4me/settings.py 
 7
 8
 9
10
11
12
13
14
15
class SqliteDatabaseSettings(BaseModel):
    """
    Settings for connections to SqLite databases.
    """

    connect_args: Optional[dict] = Field(
        default_factory=dict,
        description="Arguments for the sqlite3.connect() function.",
    )

get_engine(stg, is_async=False)

Creates an engine based on settings.

Parameters:

Name Type Description Default
stg AllDatabaseSettings

The database settings.

 required
is_async bool

Whether to create an async engine.

 False

Raises: ConfigurationError: If the database URL is not set.

Source code in db4me/connection.py 
11
12
13
14
15
16
17
18
19
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
def get_engine(
    stg: AllDatabaseSettings, is_async: bool = False
) -> Union[Engine, AsyncEngine]:
    """
    Creates an engine based on settings.

    Args:
        stg: The database settings.
        is_async: Whether to create an async engine.
    Raises:
        ConfigurationError: If the database URL is not set.
    """
    if not stg.db.url:
        raise ConfigurationError("A database connection string is required")

    # Common arguments.
    engine_args = {
        **stg.db.model_dump(),
    }

    # Dialect-specific arguments.
    if stg.db.url.startswith("sqlite"):
        engine_args.update(stg.sqlite_db.model_dump())
        engine_args["poolclass"] = SingletonThreadPool
    elif stg.db.url.startswith("postgres"):
        engine_args.update(stg.pg_db.model_dump())

    # Create the engine.
    engine: Union[Engine, AsyncEngine]
    if is_async:
        from sqlalchemy.ext.asyncio import create_async_engine

        if "pool_size" in engine_args:
            del engine_args["pool_size"]
        if "sync_alternative" in engine_args:
            del engine_args["sync_alternative"]
        engine = create_async_engine(**engine_args)
    else:
        if stg.db.sync_alternative:
            engine_args["url"] = stg.db.sync_alternative
        if "sync_alternative" in engine_args:
            del engine_args["sync_alternative"]
        engine = create_engine(**engine_args)
    return engine

Database Connection

get_engine(stg, is_async=False)

Creates an engine based on settings.

Parameters:

Name Type Description Default
stg AllDatabaseSettings

The database settings.

 required
is_async bool

Whether to create an async engine.

 False

Raises: ConfigurationError: If the database URL is not set.

Source code in db4me/connection.py 
11
12
13
14
15
16
17
18
19
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
def get_engine(
    stg: AllDatabaseSettings, is_async: bool = False
) -> Union[Engine, AsyncEngine]:
    """
    Creates an engine based on settings.

    Args:
        stg: The database settings.
        is_async: Whether to create an async engine.
    Raises:
        ConfigurationError: If the database URL is not set.
    """
    if not stg.db.url:
        raise ConfigurationError("A database connection string is required")

    # Common arguments.
    engine_args = {
        **stg.db.model_dump(),
    }

    # Dialect-specific arguments.
    if stg.db.url.startswith("sqlite"):
        engine_args.update(stg.sqlite_db.model_dump())
        engine_args["poolclass"] = SingletonThreadPool
    elif stg.db.url.startswith("postgres"):
        engine_args.update(stg.pg_db.model_dump())

    # Create the engine.
    engine: Union[Engine, AsyncEngine]
    if is_async:
        from sqlalchemy.ext.asyncio import create_async_engine

        if "pool_size" in engine_args:
            del engine_args["pool_size"]
        if "sync_alternative" in engine_args:
            del engine_args["sync_alternative"]
        engine = create_async_engine(**engine_args)
    else:
        if stg.db.sync_alternative:
            engine_args["url"] = stg.db.sync_alternative
        if "sync_alternative" in engine_args:
            del engine_args["sync_alternative"]
        engine = create_engine(**engine_args)
    return engine

Settings

Pydantic models used in configuring the database.

AllDatabaseSettings

Bases: BaseModel

Settings for the database connection.

Source code in db4me/settings.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class AllDatabaseSettings(BaseModel):
    """
    Settings for the database connection.
    """

    db: DatabaseSettings = Field(
        default_factory=cast(Callable[[], Any], DatabaseSettings),
        description="Database settings.",
    )
    pg_db: PgDatabaseSettings = Field(
        default_factory=PgDatabaseSettings,
        description="Database settings specific for postgres.",
    )
    sqlite_db: SqliteDatabaseSettings = Field(
        default_factory=SqliteDatabaseSettings,
        description="Database settings specific for SqLite.",
    )

DatabaseSettings

Bases: BaseModel

Common settings for all database types.

Attributes:

Name Type Description
url Optional[str]

The database URL. In case of async connections this is where the async database URL should be set.
sync_alternative Optional[str]

The database URL to use for sync connections. This should be used when an async connection exists in url. If missing an attempt will be made to convert the async URL to a sync URL.
echo Optional[bool]

If True, the engine will log all statements as well as a repr() of their parameter lists to the engines logger, which defaults to sys.stdout.
pool_size Optional[int]

The size of the database pool.
pool_recycle Optional[int]

The number of seconds after which a database connection should be recycled.
pool_pre_ping Optional[bool]

If True, the database connection pool will be pre-pinged This means that the connection will be checked if it is still alive before it is used.
pool_reset_on_return Optional[str]

The action to take when a connection is returned to the pool. Possible values are: - "rollback": Rollback the transaction. - "commit": Commit the transaction. - "disconnect": Disconnect the connection.
Source code in db4me/settings.py 
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
59
60
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
class DatabaseSettings(BaseModel):
    """
    Common settings for all database types.

    Attributes:
        url: The database URL. In case of async connections this is where the
            async database URL should be set.
        sync_alternative: The database URL to use for sync connections. This
            should be used when an async connection exists in `url`. If missing
            an attempt will be made to convert the async URL to a sync URL.
        echo: If True, the engine will log all statements as well as a
            repr() of their parameter lists to the engines logger, which
            defaults to sys.stdout.
        pool_size: The size of the database pool.
        pool_recycle: The number of seconds after which a database connection
            should be recycled.
        pool_pre_ping: If True, the database connection pool will be pre-pinged
            This means that the connection will be checked if it is still
            alive before it is used.
        pool_reset_on_return: The action to take when a connection is returned
            to the pool. Possible values are:
                - "rollback": Rollback the transaction.
                - "commit": Commit the transaction.
                - "disconnect": Disconnect the connection.
    """

    # Database URL
    url: Optional[str] = "sqlite://"

    # Database alternative URL.
    sync_alternative: Optional[str] = Field(None, validate_default=True)

    # Database echo
    echo: Optional[bool] = False

    # Database pool size
    pool_size: Optional[int] = 10

    # Database pool recycle
    pool_recycle: Optional[int] = 3600

    # Database pool pre ping
    pool_pre_ping: Optional[bool] = True

    # Database pool reset on return
    pool_reset_on_return: Optional[str] = "rollback"

    @field_validator("sync_alternative")
    @classmethod
    def validate_sync_alternative(cls, v, values: ValidationInfo):
        """
        Validate the sync_alternative field.
        """
        url = values.data.get("url")
        if v is None and url is not None:
            if url.startswith("postgresql+asyncpg"):
                v = url.replace("postgresql+asyncpg", "postgresql")
            elif url.startswith("sqlite+aiosqlite"):
                v = url.replace("sqlite+aiosqlite", "sqlite")
        return v

validate_sync_alternative(v, values) classmethod

Validate the sync_alternative field.

Source code in db4me/settings.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
@field_validator("sync_alternative")
@classmethod
def validate_sync_alternative(cls, v, values: ValidationInfo):
    """
    Validate the sync_alternative field.
    """
    url = values.data.get("url")
    if v is None and url is not None:
        if url.startswith("postgresql+asyncpg"):
            v = url.replace("postgresql+asyncpg", "postgresql")
        elif url.startswith("sqlite+aiosqlite"):
            v = url.replace("sqlite+aiosqlite", "sqlite")
    return v

PgDatabaseSettings

Bases: BaseModel

Settings for connections to postgres databases.

Source code in db4me/settings.py 
18
19
20
21
22
23
class PgDatabaseSettings(BaseModel):
    """
    Settings for connections to postgres databases.
    """

    pass

SqliteDatabaseSettings

Bases: BaseModel

Settings for connections to SqLite databases.

Source code in db4me/settings.py 
 7
 8
 9
10
11
12
13
14
15
class SqliteDatabaseSettings(BaseModel):
    """
    Settings for connections to SqLite databases.
    """

    connect_args: Optional[dict] = Field(
        default_factory=dict,
        description="Arguments for the sqlite3.connect() function.",
    )

Errors

ConfigurationError

Bases: Db4MeError

Raised when there is an error in the configuration file.

Source code in db4me/errors.py 
 7
 8
 9
10
class ConfigurationError(Db4MeError):
    """Raised when there is an error in the configuration file."""

    pass

Db4MeError

Bases: Exception

Base class for exceptions in this module.

Source code in db4me/errors.py 
1
2
3
4
class Db4MeError(Exception):
    """Base class for exceptions in this module."""

    pass