mirror of
https://github.com/dat515-2025/Group-8.git
synced 2026-03-22 15:12:08 +01:00
53 lines
1.9 KiB
Python
53 lines
1.9 KiB
Python
import os
|
|
from typing import AsyncGenerator
|
|
|
|
from fastapi import Depends
|
|
from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase
|
|
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
|
from sqlalchemy.ext.declarative import DeclarativeMeta, declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
DATABASE_URL = os.getenv("DATABASE_URL")
|
|
if not DATABASE_URL:
|
|
mariadb_host = os.getenv("MARIADB_HOST", "localhost")
|
|
mariadb_port = os.getenv("MARIADB_PORT", "3306")
|
|
mariadb_db = os.getenv("MARIADB_DB", "group_project")
|
|
mariadb_user = os.getenv("MARIADB_USER", "root")
|
|
mariadb_password = os.getenv("MARIADB_PASSWORD", "strongpassword")
|
|
#always use SSL except for localhost - i dont want to include certs
|
|
ssl_param = "?ssl=true" if mariadb_host != "localhost" else ""
|
|
|
|
if mariadb_host and mariadb_db and mariadb_user and mariadb_password:
|
|
# Use MariaDB/MySQL over async driver
|
|
DATABASE_URL = f"mysql+asyncmy://{mariadb_user}:{mariadb_password}@{mariadb_host}:{mariadb_port}/{mariadb_db}{ssl_param}"
|
|
else:
|
|
raise Exception("Only MariaDB is supported. Please set the DATABASE_URL environment variable.")
|
|
|
|
Base: DeclarativeMeta = declarative_base()
|
|
|
|
|
|
class User(SQLAlchemyBaseUserTableUUID, Base):
|
|
pass
|
|
|
|
|
|
engine = create_async_engine(
|
|
DATABASE_URL,
|
|
pool_pre_ping=True,
|
|
echo=os.getenv("SQL_ECHO", "0") == "1",
|
|
)
|
|
async_session_maker = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
|
|
async def create_db_and_tables():
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
|
|
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
|
|
async with async_session_maker() as session:
|
|
yield session
|
|
|
|
|
|
async def get_user_db(session: AsyncSession = Depends(get_async_session)):
|
|
yield SQLAlchemyUserDatabase(session, User)
|