Set up database layer for todo app

- Created app/db/base.py with SQLAlchemy Base to avoid circular imports
- Created app/db/session.py with SQLite database connection using /app/storage/db path
- Created app/models/todo.py with Todo model including all required fields
- Created app/schemas/todo.py with Pydantic schemas for request/response
- Added requirements.txt with FastAPI, SQLAlchemy, and other dependencies
- Created proper package structure with __init__.py files
This commit is contained in:
Automated Action 2025-06-20 23:17:54 +00:00
parent 06ebd10418
commit 16000f8745
14 changed files with 230 additions and 0 deletions

0
app/__init__.py Normal file
View File

0
app/api/__init__.py Normal file
View File

8
app/api/api.py Normal file
View File

@ -0,0 +1,8 @@
from fastapi import APIRouter
from app.api.endpoints import todos
api_router = APIRouter()
# Include todos router
api_router.include_router(todos.router, prefix="/todos", tags=["todos"])

View File

@ -0,0 +1 @@
# API endpoints package

View File

@ -0,0 +1,83 @@
from typing import List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.models.todo import Todo
from app.schemas.todo import TodoCreate, TodoUpdate, TodoResponse
router = APIRouter()
@router.get("/", response_model=List[TodoResponse])
def get_todos(db: Session = Depends(get_db)):
"""
Get all todos
"""
todos = db.query(Todo).all()
return todos
@router.post("/", response_model=TodoResponse, status_code=status.HTTP_201_CREATED)
def create_todo(todo: TodoCreate, db: Session = Depends(get_db)):
"""
Create a new todo
"""
db_todo = Todo(**todo.dict())
db.add(db_todo)
db.commit()
db.refresh(db_todo)
return db_todo
@router.get("/{todo_id}", response_model=TodoResponse)
def get_todo(todo_id: int, db: Session = Depends(get_db)):
"""
Get a specific todo by ID
"""
todo = db.query(Todo).filter(Todo.id == todo_id).first()
if not todo:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Todo with id {todo_id} not found"
)
return todo
@router.put("/{todo_id}", response_model=TodoResponse)
def update_todo(todo_id: int, todo_update: TodoUpdate, db: Session = Depends(get_db)):
"""
Update a specific todo by ID
"""
todo = db.query(Todo).filter(Todo.id == todo_id).first()
if not todo:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Todo with id {todo_id} not found"
)
# Update only the fields that are provided
update_data = todo_update.dict(exclude_unset=True)
for field, value in update_data.items():
setattr(todo, field, value)
db.commit()
db.refresh(todo)
return todo
@router.delete("/{todo_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_todo(todo_id: int, db: Session = Depends(get_db)):
"""
Delete a specific todo by ID
"""
todo = db.query(Todo).filter(Todo.id == todo_id).first()
if not todo:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Todo with id {todo_id} not found"
)
db.delete(todo)
db.commit()
return None

0
app/db/__init__.py Normal file
View File

3
app/db/base.py Normal file
View File

@ -0,0 +1,3 @@
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

25
app/db/session.py Normal file
View File

@ -0,0 +1,25 @@
from pathlib import Path
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# Database configuration using absolute path as specified
DB_DIR = Path("/app") / "storage" / "db"
DB_DIR.mkdir(parents=True, exist_ok=True)
SQLALCHEMY_DATABASE_URL = f"sqlite:///{DB_DIR}/db.sqlite"
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Dependency to get database session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

0
app/models/__init__.py Normal file
View File

14
app/models/todo.py Normal file
View File

@ -0,0 +1,14 @@
from datetime import datetime
from sqlalchemy import Column, Integer, String, Text, Boolean, DateTime
from app.db.base import Base
class Todo(Base):
__tablename__ = "todos"
id = Column(Integer, primary_key=True, index=True)
title = Column(String(255), nullable=False, index=True)
description = Column(Text, nullable=True)
completed = Column(Boolean, default=False, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)

0
app/schemas/__init__.py Normal file
View File

28
app/schemas/todo.py Normal file
View File

@ -0,0 +1,28 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel
class TodoBase(BaseModel):
title: str
description: Optional[str] = None
completed: bool = False
class TodoCreate(TodoBase):
pass
class TodoUpdate(BaseModel):
title: Optional[str] = None
description: Optional[str] = None
completed: Optional[bool] = None
class TodoResponse(TodoBase):
id: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True

62
main.py Normal file
View File

@ -0,0 +1,62 @@
from fastapi import FastAPI, Depends
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy.orm import Session
from app.db.session import get_db, engine
from app.db.base import Base
import os
# Create database tables
Base.metadata.create_all(bind=engine)
# Initialize FastAPI app
app = FastAPI(
title="Todo App API",
description="A simple Todo application API built with FastAPI",
version="1.0.0",
docs_url="/docs",
redoc_url="/redoc",
openapi_url="/openapi.json"
)
# CORS configuration - allow all origins
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/")
async def root():
"""Base URL endpoint with app information and links"""
return {
"title": "Todo App API",
"description": "A simple Todo application API built with FastAPI",
"version": "1.0.0",
"documentation": {
"swagger_ui": "/docs",
"redoc": "/redoc",
"openapi_json": "/openapi.json"
},
"health_check": "/health"
}
@app.get("/health")
async def health_check(db: Session = Depends(get_db)):
"""Health check endpoint that reports application health"""
try:
# Test database connection
db.execute("SELECT 1")
db_status = "healthy"
except Exception as e:
db_status = f"unhealthy: {str(e)}"
return {
"status": "healthy" if db_status == "healthy" else "unhealthy",
"database": db_status,
"service": "Todo App API",
"version": "1.0.0"
}

6
requirements.txt Normal file
View File

@ -0,0 +1,6 @@
fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
pydantic==2.5.0
alembic==1.12.1
ruff==0.1.6