Refactor code to improve typing and linting

This commit is contained in:
Automated Action 2025-05-16 00:42:00 +00:00
parent 8e26cae20e
commit fb99c09cdd
16 changed files with 68 additions and 52 deletions

View File

@ -47,7 +47,7 @@ Once the server is running, you can access:
## API Endpoints ## API Endpoints
- `GET /health` - Health check endpoint - `GET /api/health` - Health check endpoint
- `GET /api/todos` - List all todos - `GET /api/todos` - List all todos
- `GET /api/todos/{id}` - Get a single todo by ID - `GET /api/todos/{id}` - Get a single todo by ID
- `POST /api/todos` - Create a new todo - `POST /api/todos` - Create a new todo

View File

@ -1,10 +1,12 @@
from typing import Optional
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from typing import List, Optional
from db.models import Todo
from api.schemas.todo import TodoCreate, TodoUpdate from api.schemas.todo import TodoCreate, TodoUpdate
from db.models import Todo
def get_todos(db: Session, skip: int = 0, limit: int = 100) -> List[Todo]: def get_todos(db: Session, skip: int = 0, limit: int = 100) -> list[Todo]:
""" """
Get all todos with pagination Get all todos with pagination
""" """

View File

@ -1,10 +1,11 @@
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
import platform import platform
import sys import sys
from db.database import get_db from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from api.schemas.health import HealthResponse from api.schemas.health import HealthResponse
from db.database import get_db
router = APIRouter() router = APIRouter()
@ -20,7 +21,8 @@ def health_check(db: Session = Depends(get_db)):
# Just run a simple query to verify DB connection # Just run a simple query to verify DB connection
db.execute("SELECT 1") db.execute("SELECT 1")
db_status = "healthy" db_status = "healthy"
except Exception as e: except (RuntimeError, ConnectionError) as e:
# Catch specific exceptions rather than a broad Exception
db_status = f"unhealthy: {str(e)}" db_status = f"unhealthy: {str(e)}"
return HealthResponse( return HealthResponse(
@ -30,5 +32,5 @@ def health_check(db: Session = Depends(get_db)):
"database": db_status, "database": db_status,
"python_version": sys.version, "python_version": sys.version,
"platform": platform.platform(), "platform": platform.platform(),
} },
) )

View File

@ -1,15 +1,14 @@
from fastapi import APIRouter, Depends, HTTPException, status from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from typing import List
from db.database import get_db from api.crud.todo import create_todo, delete_todo, get_todo, get_todos, update_todo
from api.schemas.todo import TodoCreate, TodoResponse, TodoUpdate from api.schemas.todo import TodoCreate, TodoResponse, TodoUpdate
from api.crud.todo import get_todos, get_todo, create_todo, update_todo, delete_todo from db.database import get_db
router = APIRouter() router = APIRouter()
@router.get("/", response_model=List[TodoResponse]) @router.get("/", response_model=list[TodoResponse])
def read_todos(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): def read_todos(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
""" """
Get all todos with pagination Get all todos with pagination

View File

@ -1,9 +1,10 @@
from typing import Any
from pydantic import BaseModel from pydantic import BaseModel
from typing import Dict, Any
class HealthResponse(BaseModel): class HealthResponse(BaseModel):
"""Response schema for the health check endpoint""" """Response schema for the health check endpoint"""
status: str status: str
version: str version: str
details: Dict[str, Any] details: dict[str, Any]

View File

@ -1,12 +1,15 @@
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field
class TodoBase(BaseModel): class TodoBase(BaseModel):
"""Base Todo schema with common attributes""" """Base Todo schema with common attributes"""
title: str = Field(..., min_length=1, max_length=100, description="Title of the todo") title: str = Field(..., min_length=1, max_length=100, description="Title of the todo")
description: Optional[str] = Field(None, max_length=500, description="Detailed description of the todo") description: Optional[str] = Field(
None, max_length=500, description="Detailed description of the todo",
)
completed: bool = Field(False, description="Whether the todo is completed") completed: bool = Field(False, description="Whether the todo is completed")
@ -17,8 +20,12 @@ class TodoCreate(TodoBase):
class TodoUpdate(BaseModel): class TodoUpdate(BaseModel):
"""Schema for updating an existing todo, all fields are optional""" """Schema for updating an existing todo, all fields are optional"""
title: Optional[str] = Field(None, min_length=1, max_length=100, description="Title of the todo") title: Optional[str] = Field(
description: Optional[str] = Field(None, max_length=500, description="Detailed description of the todo") None, min_length=1, max_length=100, description="Title of the todo",
)
description: Optional[str] = Field(
None, max_length=500, description="Detailed description of the todo",
)
completed: Optional[bool] = Field(None, description="Whether the todo is completed") completed: Optional[bool] = Field(None, description="Whether the todo is completed")

View File

@ -1,4 +1,5 @@
from pathlib import Path from pathlib import Path
from sqlalchemy import create_engine from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker from sqlalchemy.orm import sessionmaker
@ -13,7 +14,7 @@ SQLALCHEMY_DATABASE_URL = f"sqlite:///{DB_DIR}/db.sqlite"
# Create engine # Create engine
engine = create_engine( engine = create_engine(
SQLALCHEMY_DATABASE_URL, SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False} connect_args={"check_same_thread": False},
) )
# Create session factory # Create session factory

View File

@ -1,5 +1,6 @@
from sqlalchemy import Column, Integer, String, Boolean, DateTime from sqlalchemy import Boolean, Column, DateTime, Integer, String
from sqlalchemy.sql import func from sqlalchemy.sql import func
from .database import Base from .database import Base

View File

@ -1,7 +1,8 @@
import uvicorn
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
import uvicorn
from api.routers import todo_router, health_router from api.routers import health_router, todo_router
from db.database import create_tables from db.database import create_tables
app = FastAPI( app = FastAPI(
@ -21,7 +22,7 @@ app.add_middleware(
# Include routers # Include routers
app.include_router(todo_router.router, prefix="/api/todos", tags=["todos"]) app.include_router(todo_router.router, prefix="/api/todos", tags=["todos"])
app.include_router(health_router.router, tags=["health"]) app.include_router(health_router.router, prefix="/api", tags=["health"])
# Create database tables on startup # Create database tables on startup
@app.on_event("startup") @app.on_event("startup")

View File

@ -1,9 +1,7 @@
from logging.config import fileConfig from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context from alembic import context
from sqlalchemy import engine_from_config, pool
from db.models import Base from db.models import Base
@ -66,7 +64,7 @@ def run_migrations_online():
with connectable.connect() as connection: with connectable.connect() as connection:
context.configure( context.configure(
connection=connection, target_metadata=target_metadata connection=connection, target_metadata=target_metadata,
) )
with context.begin_transaction(): with context.begin_transaction():

View File

@ -5,9 +5,8 @@ Revises:
Create Date: 2023-09-20 12:00:00.000000 Create Date: 2023-09-20 12:00:00.000000
""" """
from alembic import op
import sqlalchemy as sa import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic. # revision identifiers, used by Alembic.
revision = '1_initial_create_todos_table' revision = '1_initial_create_todos_table'
@ -25,7 +24,7 @@ def upgrade():
sa.Column('completed', sa.Boolean(), default=False), sa.Column('completed', sa.Boolean(), default=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()), sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.func.now()),
sa.Column('updated_at', sa.DateTime(timezone=True), onupdate=sa.func.now()), sa.Column('updated_at', sa.DateTime(timezone=True), onupdate=sa.func.now()),
sa.PrimaryKeyConstraint('id') sa.PrimaryKeyConstraint('id'),
) )
op.create_index(op.f('ix_todos_id'), 'todos', ['id'], unique=False) op.create_index(op.f('ix_todos_id'), 'todos', ['id'], unique=False)

View File

@ -1,8 +1,13 @@
[tool.ruff] [tool.ruff]
line-length = 100 line-length = 100
target-version = "py39" target-version = "py39"
select = ["E", "F", "I", "W", "N", "B", "A", "COM", "C4", "UP", "S", "BLE", "T10", "ISC", "G"]
ignore = []
[tool.ruff.isort] [tool.ruff.lint]
select = ["E", "F", "I", "W", "N", "B", "A", "COM", "C4", "UP", "S", "BLE", "T10", "ISC", "G"]
ignore = [
"B008", # Function call in argument defaults (used by FastAPI for dependency injection)
"S104", # Binding to all interfaces (common in development environments)
]
[tool.ruff.lint.isort]
known-third-party = ["fastapi", "sqlalchemy", "pydantic", "alembic"] known-third-party = ["fastapi", "sqlalchemy", "pydantic", "alembic"]