Spaces:
Runtime error
Runtime error
File size: 1,453 Bytes
6e8003f 7114487 6e8003f 7114487 6e8003f 7114487 6e8003f |
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 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 |
from typing import AsyncGenerator
from fastapi import Depends
from fastapi_users.db import SQLAlchemyBaseUserTableUUID, SQLAlchemyUserDatabase
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase
from dotenv import load_dotenv
import os
# Get the current environment from the environment variable
current_environment = os.getenv("APP_ENV", "dev")
# Load the appropriate .env file based on the current environment
if current_environment == "dev":
load_dotenv(".env.dev")
elif current_environment == "test":
load_dotenv(".env.test")
elif current_environment == "prod":
load_dotenv(".env.prod")
else:
raise ValueError("Invalid environment specified")
db_connection_string = os.getenv("DB_CONNECTION_STRING")
DATABASE_URL = db_connection_string
class Base(DeclarativeBase):
pass
class User(SQLAlchemyBaseUserTableUUID, Base):
pass
engine = create_async_engine(DATABASE_URL)
async_session_maker = async_sessionmaker(engine, 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)
|