generated from MrSphay/codex-agent-repository-kit
9800c0bfb6
Some checks failed
Build and Push Docker Image / build-and-push (push) Failing after 3m1s
87 lines
3.2 KiB
Python
87 lines
3.2 KiB
Python
import os
|
|
from fastapi import FastAPI, Request, Form, Depends, HTTPException, Response
|
|
from fastapi.responses import RedirectResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from sqlalchemy import create_engine, Column, Integer, String, Float
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
|
|
# Database Setup
|
|
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./data/contracts.db")
|
|
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False} if "sqlite" in DATABASE_URL else {})
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
Base = declarative_base()
|
|
|
|
class Contract(Base):
|
|
__tablename__ = "contracts"
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
provider = Column(String)
|
|
phone_number = Column(String)
|
|
pin = Column(String)
|
|
puk = Column(String)
|
|
monthly_cost = Column(Float)
|
|
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
app = FastAPI()
|
|
templates = Jinja2Templates(directory="templates")
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
@app.get("/")
|
|
def list_contracts(request: Request, db: Session = Depends(get_db)):
|
|
contracts = db.query(Contract).all()
|
|
return templates.TemplateResponse("index.html", {"request": request, "contracts": contracts})
|
|
|
|
@app.get("/add")
|
|
def add_form(request: Request):
|
|
return templates.TemplateResponse("form.html", {"request": request, "action": "add"})
|
|
|
|
@app.post("/add")
|
|
def add_contract(
|
|
provider: str = Form(...),
|
|
phone_number: str = Form(...),
|
|
pin: str = Form(...),
|
|
puk: str = Form(...),
|
|
monthly_cost: float = Form(...),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
new_contract = Contract(
|
|
provider=provider, phone_number=phone_number,
|
|
pin=pin, puk=puk, monthly_cost=monthly_cost
|
|
)
|
|
db.add(new_contract)
|
|
db.commit()
|
|
return RedirectResponse(url="/", status_code=303)
|
|
|
|
@app.get("/edit/{contract_id}")
|
|
def edit_form(request: Request, contract_id: int, db: Session = Depends(get_db)):
|
|
contract = db.query(Contract).filter(Contract.id == contract_id).first()
|
|
if not contract: raise HTTPException(status_code=404)
|
|
return templates.TemplateResponse("form.html", {"request": request, "contract": contract, "action": "edit"})
|
|
|
|
@app.post("/edit/{contract_id}")
|
|
def edit_contract(
|
|
contract_id: int, provider: str = Form(...), phone_number: str = Form(...),
|
|
pin: str = Form(...), puk: str = Form(...), monthly_cost: float = Form(...),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
contract = db.query(Contract).filter(Contract.id == contract_id).first()
|
|
if not contract: raise HTTPException(status_code=404)
|
|
contract.provider, contract.phone_number, contract.pin, contract.puk, contract.monthly_cost = provider, phone_number, pin, puk, monthly_cost
|
|
db.commit()
|
|
return RedirectResponse(url="/", status_code=303)
|
|
|
|
@app.get("/delete/{contract_id}")
|
|
def delete_contract(contract_id: int, db: Session = Depends(get_db)):
|
|
contract = db.query(Contract).filter(Contract.id == contract_id).first()
|
|
if contract:
|
|
db.delete(contract)
|
|
db.commit()
|
|
return RedirectResponse(url="/", status_code=303)
|