121 lines
3.2 KiB
Python
121 lines
3.2 KiB
Python
from fastapi import FastAPI, HTTPException, Depends, status
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import create_engine, Column, Integer, String
|
|
from sqlalchemy.ext.declarative import declarative_base
|
|
from sqlalchemy.orm import sessionmaker, Session
|
|
from typing import List, Optional
|
|
from contextlib import asynccontextmanager
|
|
import os
|
|
|
|
|
|
# Database setup
|
|
SQLALCHEMY_DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///./team_status.db")
|
|
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
|
|
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
Base = declarative_base()
|
|
|
|
|
|
# SQLAlchemy Model
|
|
class Employee(Base):
|
|
__tablename__ = "employees"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
name = Column(String, nullable=False)
|
|
position = Column(String, nullable=False)
|
|
status = Column(String, default="Online")
|
|
|
|
|
|
# Create tables
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
|
|
# Pydantic Models
|
|
class EmployeeBase(BaseModel):
|
|
name: str
|
|
position: str
|
|
|
|
|
|
class EmployeeCreate(EmployeeBase):
|
|
pass
|
|
|
|
|
|
class EmployeeResponse(EmployeeBase):
|
|
id: int
|
|
status: str
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class StatusUpdate(BaseModel):
|
|
status: str
|
|
|
|
|
|
# FastAPI app
|
|
app = FastAPI(title="Team Status Board API")
|
|
|
|
# CORS middleware
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
# Dependency
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
# CRUD Operations
|
|
@app.get("/api/employees", response_model=List[EmployeeResponse])
|
|
def read_employees(db: Session = Depends(get_db)):
|
|
"""Get all employees"""
|
|
employees = db.query(Employee).all()
|
|
return employees
|
|
|
|
|
|
@app.post("/api/employees", response_model=EmployeeResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_employee(employee: EmployeeCreate, db: Session = Depends(get_db)):
|
|
"""Create a new employee"""
|
|
db_employee = Employee(**employee.model_dump(), status="Online")
|
|
db.add(db_employee)
|
|
db.commit()
|
|
db.refresh(db_employee)
|
|
return db_employee
|
|
|
|
|
|
@app.put("/api/employees/{employee_id}/status", response_model=EmployeeResponse)
|
|
def update_employee_status(employee_id: int, status_update: StatusUpdate, db: Session = Depends(get_db)):
|
|
"""Update employee status"""
|
|
employee = db.query(Employee).filter(Employee.id == employee_id).first()
|
|
if not employee:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Employee not found")
|
|
employee.status = status_update.status
|
|
db.commit()
|
|
db.refresh(employee)
|
|
return employee
|
|
|
|
|
|
@app.delete("/api/employees/{employee_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_employee(employee_id: int, db: Session = Depends(get_db)):
|
|
"""Delete an employee"""
|
|
employee = db.query(Employee).filter(Employee.id == employee_id).first()
|
|
if not employee:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Employee not found")
|
|
db.delete(employee)
|
|
db.commit()
|
|
return None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|