1
2
3
4
5
6
7
8
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
|
import os
import ormar
import sqlalchemy
from ormar.databases.connection import DatabaseConnection
from sqlalchemy.ext.asyncio import create_async_engine
def convert_to_async_url(url: str) -> str: # pragma: nocover
"""Convert database URL to async driver version."""
if url.startswith("postgresql://"):
return url.replace("postgresql://", "postgresql+asyncpg://", 1)
elif url.startswith("mysql://"):
return url.replace("mysql://", "mysql+aiomysql://", 1)
elif url.startswith("sqlite://"):
return url.replace("sqlite://", "sqlite+aiosqlite://", 1)
return url
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///test.db")
ASYNC_DATABASE_URL = convert_to_async_url(DATABASE_URL)
print("USED DB:", ASYNC_DATABASE_URL)
def create_config(**args):
database_ = DatabaseConnection(ASYNC_DATABASE_URL, **args)
metadata_ = sqlalchemy.MetaData()
async_engine_ = create_async_engine(ASYNC_DATABASE_URL)
return ormar.OrmarConfig(
metadata=metadata_,
database=database_,
engine=async_engine_,
)
|
