16d63c4168
- Set up FastAPI application with CORS support - Implement SQLite database with SQLAlchemy ORM - Create User model with CRUD operations - Add Alembic for database migrations - Include health check and documentation endpoints - Set up proper project structure with organized modules - Add comprehensive README with setup instructions - Configure Ruff for code linting and formatting
99 lines
2.7 KiB
Python
99 lines
2.7 KiB
Python
from fastapi import FastAPI, HTTPException, Depends
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db.session import SessionLocal, engine
|
|
from app.db.base import Base
|
|
from app.schemas.user import UserCreate, UserResponse
|
|
from app.crud.user import create_user, get_user, get_users, update_user, delete_user
|
|
|
|
# Create database tables
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
app = FastAPI(
|
|
title="REST API Service",
|
|
description="A REST API service built with FastAPI and SQLite",
|
|
version="1.0.0",
|
|
docs_url="/docs",
|
|
redoc_url="/redoc",
|
|
openapi_url="/openapi.json",
|
|
)
|
|
|
|
# CORS middleware
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
|
|
# Dependency to get database session
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return {
|
|
"title": "REST API Service",
|
|
"description": "A REST API service built with FastAPI and SQLite",
|
|
"documentation": "/docs",
|
|
"health_check": "/health",
|
|
}
|
|
|
|
|
|
@app.get("/health")
|
|
async def health_check():
|
|
return {"status": "healthy", "service": "REST API Service"}
|
|
|
|
|
|
# User endpoints
|
|
@app.post("/users/", response_model=UserResponse)
|
|
async def create_user_endpoint(user: UserCreate, db: Session = Depends(get_db)):
|
|
db_user = create_user(db=db, user=user)
|
|
return db_user
|
|
|
|
|
|
@app.get("/users/", response_model=list[UserResponse])
|
|
async def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
|
|
users = get_users(db, skip=skip, limit=limit)
|
|
return users
|
|
|
|
|
|
@app.get("/users/{user_id}", response_model=UserResponse)
|
|
async def read_user(user_id: int, db: Session = Depends(get_db)):
|
|
db_user = get_user(db, user_id=user_id)
|
|
if db_user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return db_user
|
|
|
|
|
|
@app.put("/users/{user_id}", response_model=UserResponse)
|
|
async def update_user_endpoint(
|
|
user_id: int, user: UserCreate, db: Session = Depends(get_db)
|
|
):
|
|
db_user = update_user(db=db, user_id=user_id, user=user)
|
|
if db_user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return db_user
|
|
|
|
|
|
@app.delete("/users/{user_id}")
|
|
async def delete_user_endpoint(user_id: int, db: Session = Depends(get_db)):
|
|
success = delete_user(db=db, user_id=user_id)
|
|
if not success:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
return {"message": "User deleted successfully"}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
|
|
uvicorn.run(app, host="0.0.0.0", port=8000)
|