44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
from __future__ import annotations
|
|
import os
|
|
from contextlib import asynccontextmanager
|
|
from typing import AsyncGenerator
|
|
from dotenv import load_dotenv
|
|
from qdrant_client import AsyncQdrantClient, models
|
|
from typing import Annotated
|
|
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
|
|
from sqlalchemy.orm import declarative_base
|
|
from sqlalchemy.orm import sessionmaker
|
|
# pyrefly: ignore [missing-import]
|
|
from fastapi import FastAPI, Depends
|
|
from sqlalchemy.engine.url import URL
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
|
|
DATABASE_URL = URL.create(
|
|
drivername="mysql+asyncmy",
|
|
username=os.getenv("MYSQL_USER"),
|
|
password=os.getenv("MYSQL_PASSWORD"),
|
|
host=os.getenv("MYSQL_HOST"),
|
|
port=os.getenv("MYSQL_PORT"),
|
|
database=os.getenv("MYSQL_DATABASE"),
|
|
)
|
|
|
|
|
|
async def get_qdrant_client()->AsyncGenerator[AsyncQdrantClient,None]:
|
|
# Replace with your Qdrant URL
|
|
client = AsyncQdrantClient(url="http://localhost:6333", timeout=60)
|
|
try:
|
|
yield client
|
|
finally:
|
|
# Properly close the async client
|
|
await client.close()
|
|
|
|
async def get_session():
|
|
engine = create_async_engine(DATABASE_URL, echo=True)
|
|
async_session = async_sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
|
|
session = async_session()
|
|
Base = declarative_base()
|
|
try:
|
|
yield session
|
|
finally:
|
|
await session.close() |