38 lines
978 B
Python
38 lines
978 B
Python
from fastapi import FastAPI
|
|
import asyncmy
|
|
import asyncio
|
|
import os
|
|
from contextlib import asynccontextmanager
|
|
|
|
@asynccontextmanager
|
|
async def connect_db(app: FastAPI):
|
|
app.state.pool = await asyncmy.create_pool(
|
|
host=os.getenv("DB_HOST", "localhost"),
|
|
port=int(os.getenv("DB_PORT", 3306)),
|
|
user=os.getenv("DB_USER"),
|
|
password=os.getenv("DB_PASSWORD"),
|
|
db=os.getenv("DB_NAME"),
|
|
minsize=5,
|
|
maxsize=20
|
|
)
|
|
try:
|
|
yield
|
|
finally:
|
|
app.state.pool.close()
|
|
await app.state.pool.wait_closed()
|
|
|
|
app = FastAPI(lifespan=connect_db)
|
|
|
|
@asynccontextmanager
|
|
async def get_connection():
|
|
async with app.state.pool.acquire() as conn:
|
|
yield conn
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "yes the api works"}
|
|
|
|
@app.get("/random")
|
|
async def get_random_bnuy():
|
|
async with get_connection() as conn:
|
|
return {"message": "here could be a bnuy, if I would've implemented it"} |