80 lines
2.0 KiB
Python
80 lines
2.0 KiB
Python
import time
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy.exc import OperationalError
|
|
from starlette.middleware.sessions import SessionMiddleware
|
|
|
|
from app.config import get_settings
|
|
from app.database import Base, engine
|
|
from app.routers import admin as admin_router
|
|
from app.routers import shop as shop_router
|
|
from app.seed import seed_if_empty
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
settings = get_settings()
|
|
|
|
|
|
def wait_for_db(retries: int = 30, delay: float = 1.0) -> None:
|
|
for attempt in range(1, retries + 1):
|
|
try:
|
|
with engine.connect() as conn:
|
|
conn.exec_driver_sql("SELECT 1")
|
|
return
|
|
except OperationalError:
|
|
if attempt == retries:
|
|
raise
|
|
time.sleep(delay)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI):
|
|
wait_for_db()
|
|
Base.metadata.create_all(bind=engine)
|
|
from app.database import SessionLocal
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
seed_if_empty(db)
|
|
finally:
|
|
db.close()
|
|
yield
|
|
|
|
|
|
app = FastAPI(title=settings.app_name, lifespan=lifespan)
|
|
app.add_middleware(
|
|
SessionMiddleware,
|
|
secret_key=settings.secret_key,
|
|
session_cookie="pitopn_session",
|
|
same_site="lax",
|
|
https_only=False,
|
|
)
|
|
app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static")
|
|
|
|
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
|
|
|
|
|
|
def format_price(value) -> str:
|
|
try:
|
|
amount = float(value)
|
|
except (TypeError, ValueError):
|
|
return str(value)
|
|
formatted = f"{amount:,.0f}".replace(",", " ")
|
|
return f"{formatted} ₽"
|
|
|
|
|
|
templates.env.filters["rub"] = format_price
|
|
|
|
shop_router.setup(templates)
|
|
admin_router.setup(templates)
|
|
app.include_router(shop_router.router)
|
|
app.include_router(admin_router.router)
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> dict[str, str]:
|
|
return {"status": "ok"}
|