Bases: BaseModel
Database connection settings.
.. note::
You can pass here any extra option supported by
`SQLAlchemy Engine class <https://docs.sqlalchemy.org/en/20/core/engines.html#sqlalchemy.create_engine>`_,
even if it is not mentioned in documentation.
Examples
.. code-block:: yaml
:caption: config.yml
database:
url: postgresql+asyncpg://postgres:postgres@localhost:5432/syncmaster
# custom option passed directly to SQLAlchemy Engine
pool_pre_ping: True
Source code in syncmaster/settings/database.py
9
10
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 | class DatabaseSettings(BaseModel):
"""Database connection settings.
.. note::
You can pass here any extra option supported by
`SQLAlchemy Engine class <https://docs.sqlalchemy.org/en/20/core/engines.html#sqlalchemy.create_engine>`_,
even if it is not mentioned in documentation.
Examples
--------
.. code-block:: yaml
:caption: config.yml
database:
url: postgresql+asyncpg://postgres:postgres@localhost:5432/syncmaster
# custom option passed directly to SQLAlchemy Engine
pool_pre_ping: True
"""
url: str = Field(
description=textwrap.dedent(
"""
Database connection URL.
See `SQLAlchemy documentation <https://docs.sqlalchemy.org/en/20/core/engines.html#server-specific-urls>`_
.. warning:
Only async drivers are supported, e.g. ``asyncpg``
""",
),
)
@property
def sync_url(self) -> str:
parsed_url = urlparse(self.url)
# replace '+asyncpg' with '+psycopg2' in the scheme - used by celery
scheme = parsed_url.scheme.replace("+asyncpg", "+psycopg2")
sync_parsed_url = parsed_url._replace(scheme=scheme)
return urlunparse(sync_parsed_url)
model_config = ConfigDict(extra="allow")
|