Ross
136 lines
3.8 KiB
Python
136 lines
3.8 KiB
Python
import uvicorn
|
|
from pydantic import BaseModel
|
|
|
|
from fastapi_sqlalchemy import DBSessionMiddleware, db
|
|
from fastapi import Depends, FastAPI, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
|
|
import os
|
|
from dotenv import load_dotenv
|
|
|
|
from rota_generator.shifts import SingleShift, ShiftConstraint
|
|
|
|
from models import SingleShift as ModelSingleShift
|
|
#from models import ShiftConstraint as ModelShiftConstraint
|
|
|
|
from database import SessionLocal, engine
|
|
|
|
import crud, models, schemas
|
|
|
|
#from rota.wokers
|
|
|
|
|
|
load_dotenv('../.env')
|
|
|
|
app = FastAPI()
|
|
|
|
# Dependency
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
#app.add_middleware(DBSessionMiddleware, db_url="sqlite:///test.db")
|
|
models.Base.metadata.create_all(bind=engine)
|
|
|
|
@app.post("/shifts/", response_model=schemas.SingleShift)
|
|
def create_shift(shift: schemas.SingleShiftCreate, db: Session = Depends(get_db)):
|
|
db_shift = crud.get_shift_by_name(db, name=shift.name)
|
|
if db_shift:
|
|
raise HTTPException(status_code=400, detail="Shift name already exists")
|
|
return crud.create_shift(db=db, shift=shift)
|
|
|
|
|
|
@app.get("/shifts/", response_model=list[schemas.SingleShift])
|
|
def read_shifts(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
shifts = crud.get_shifts(db, skip=skip, limit=limit)
|
|
return shifts
|
|
|
|
|
|
@app.get("/shifts/{shift_id}", response_model=schemas.SingleShift)
|
|
def read_shift(shift_id: int, db: Session = Depends(get_db)):
|
|
db_shift = crud.get_shifts(db, shift_id=shift_id)
|
|
if db_shift is None:
|
|
raise HTTPException(status_code=404, detail="Shift not found")
|
|
return db_shift
|
|
|
|
@app.post("/workers/", response_model=schemas.Worker)
|
|
def create_worker(worker: schemas.WorkerCreate, db: Session = Depends(get_db)):
|
|
db_worker = crud.get_worker_by_name(db, name=worker.name)
|
|
if db_worker:
|
|
raise HTTPException(status_code=400, detail="worker name already exists")
|
|
return crud.create_worker(db=db, worker=worker)
|
|
|
|
|
|
@app.get("/workers/", response_model=list[schemas.Worker])
|
|
def read_workers(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
workers = crud.get_workers(db, skip=skip, limit=limit)
|
|
return workers
|
|
|
|
|
|
@app.get("/workers/{worker_id}", response_model=schemas.Worker)
|
|
def read_worker(worker_id: int, db: Session = Depends(get_db)):
|
|
db_worker = crud.get_workers(db, worker_id=worker_id)
|
|
if db_worker is None:
|
|
raise HTTPException(status_code=404, detail="worker not found")
|
|
return db_worker
|
|
|
|
#@app.post("/users/{user_id}/items/", response_model=schemas.Item)
|
|
#def create_item_for_user(
|
|
# user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
|
|
#):
|
|
# return crud.create_user_item(db=db, item=item, user_id=user_id)
|
|
#
|
|
#
|
|
#@app.get("/items/", response_model=list[schemas.Item])
|
|
#def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
# items = crud.get_items(db, skip=skip, limit=limit)
|
|
# return items
|
|
|
|
#@app.post("/shifts/")
|
|
#async def create_shift(shift: SingleShift):
|
|
# return shift
|
|
|
|
#@app.post('/shifts/', response_model=SingleShift)
|
|
#async def shifts(shift: ModelSingleShift):
|
|
# db_shift = ModelSingleShift(name=shift.name, length=shift.length)
|
|
# db.session.add(db_shift)
|
|
# db.session.commit()
|
|
# return db_shift
|
|
|
|
|
|
#@app.get("/shifts/")
|
|
#async def shifts():
|
|
# shifts = db.session.query(ModelSingleShift).all()
|
|
#
|
|
# return shifts
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {"message": "Hello World"}
|
|
|
|
@app.get("/files/{file_path:path}")
|
|
async def read_file(file_path: str):
|
|
return {"file_path": file_path}
|
|
|
|
|
|
|
|
class Item(BaseModel):
|
|
name: str
|
|
description: str | None = None
|
|
price: float
|
|
tax: float | None = None
|
|
|
|
|
|
|
|
@app.post("/items/")
|
|
async def create_item(item: Item):
|
|
item.name + item.price
|
|
return item
|
|
|
|
if __name__ == '__main__':
|
|
uvicorn.run(app, host='0.0.0.0', port=8008) |