@@ -0,0 +1,38 @@
|
||||
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"}
|
||||
Reference in New Issue
Block a user