Add Pitopn digital shop with FastAPI, Postgres 17, and Docker Compose
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
+97
@@ -0,0 +1,97 @@
|
||||
import time
|
||||
from contextlib import asynccontextmanager
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import Depends, FastAPI, Request
|
||||
from fastapi.responses import HTMLResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from fastapi.templating import Jinja2Templates
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.exc import OperationalError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.config import get_settings
|
||||
from app.database import Base, engine, get_db
|
||||
from app.models import Category, Product
|
||||
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.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
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
def health() -> dict[str, str]:
|
||||
return {"status": "ok"}
|
||||
|
||||
|
||||
@app.get("/", response_class=HTMLResponse)
|
||||
def home(request: Request, db: Session = Depends(get_db)):
|
||||
categories = db.scalars(
|
||||
select(Category).order_by(Category.sort_order, Category.name)
|
||||
).all()
|
||||
featured = db.scalars(
|
||||
select(Product)
|
||||
.where(Product.is_active.is_(True), Product.is_featured.is_(True))
|
||||
.order_by(Product.id)
|
||||
.limit(4)
|
||||
).all()
|
||||
products = db.scalars(
|
||||
select(Product)
|
||||
.where(Product.is_active.is_(True))
|
||||
.order_by(Product.is_featured.desc(), Product.id)
|
||||
.limit(12)
|
||||
).all()
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"index.html",
|
||||
{
|
||||
"app_name": settings.app_name,
|
||||
"categories": categories,
|
||||
"featured": featured,
|
||||
"products": products,
|
||||
},
|
||||
)
|
||||
Reference in New Issue
Block a user