48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
import json
|
|
from urllib.parse import quote_plus
|
|
|
|
import pydantic
|
|
from sqlalchemy.engine import URL
|
|
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
|
|
|
|
import src.settings as settings
|
|
|
|
engine: AsyncEngine
|
|
|
|
|
|
def _custom_json_serializer(*args, **kwargs) -> str:
|
|
"""
|
|
Encodes json in the same way that pydantic does.
|
|
"""
|
|
return json.dumps(*args, default=pydantic.json.pydantic_encoder, **kwargs)
|
|
|
|
print(settings.DB_USE_SQLITE)
|
|
# check if the engine must be created for the tests -> in memory sqlite
|
|
if settings.TEST_MODE:
|
|
engine = create_async_engine(
|
|
"sqlite+aiosqlite://",
|
|
connect_args={"check_same_thread": False},
|
|
json_serializer=_custom_json_serializer,
|
|
)
|
|
elif settings.DB_USE_SQLITE:
|
|
engine = create_async_engine(
|
|
URL.create(drivername="sqlite+aiosqlite", database=settings.DB_SQLITE_PATH),
|
|
connect_args={"check_same_thread": False},
|
|
json_serializer=_custom_json_serializer,
|
|
)
|
|
else:
|
|
# use the postgresql database
|
|
_encoded_password = quote_plus(settings.DB_PASSWORD)
|
|
engine = create_async_engine(
|
|
URL.create(
|
|
drivername="postgresql+asyncpg",
|
|
username=settings.DB_USER,
|
|
password=_encoded_password,
|
|
host=settings.DB_HOST,
|
|
port=settings.DB_PORT,
|
|
database=settings.DB_NAME,
|
|
),
|
|
json_serializer=_custom_json_serializer,
|
|
pool_pre_ping=True,
|
|
)
|