File size: 3,160 Bytes
25324d4 ba86161 4f9c4cb ab4ecf2 4f9c4cb ba86161 b5eec84 ba86161 a3d1ff3 ba86161 f00b651 689ef1d 0d1b7fb e2f7936 ba86161 70f1a99 25324d4 70f1a99 25324d4 48dd949 25324d4 bbbcb70 25324d4 bbbcb70 25324d4 ab6f064 25324d4 ab6f064 25324d4 7f960b1 |
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 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
"""
from typing import AsyncGenerator
from app import app
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 sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy import URL
db_url = URL.create(
"postgresql+asyncpg",
username="avnadmin",
password="AVNS_u12fHxoNLBbzD8TGpqN", # plain (unescaped) text
host="pg-opengenai-opengenai.b.aivencloud.com",
database="defaultdb",
port=14535,
)
DATABASE_URL = db_url
#DATABASE_URL = "sqlite+aiosqlite:////./test.db"
class Base(DeclarativeBase):
pass
class User(SQLAlchemyBaseUserTableUUID, Base):
pass
engine = create_async_engine(DATABASE_URL, echo=True)
async_session_maker = sessionmaker(engine,
class_=AsyncSession,
autoflush=True,
autocommit=False,
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)
async def connect() -> None:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all, checkfirst=True)
async def disconnect() -> None:
if engine:
await engine.dispose()
"""
from typing import AsyncGenerator
from sqlalchemy import URL
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
#DATABASE_URL = "sqlite+aiosqlite:///./test.db"
db_url = URL.create(
"postgresql+asyncpg",
username="avnadmin",
password="AVNS_u12fHxoNLBbzD8TGpqN", # plain (unescaped) text
host="pg-opengenai-opengenai.b.aivencloud.com",
database="defaultdb",
port=14535,
)
DATABASE_URL = db_url
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)
async def connect() -> None:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all, checkfirst=True)
async def disconnect() -> None:
if engine:
await engine.dispose() |