Create FastAPI REST API with SQLite database

README.md

@@ -1,48 +1,107 @@
 # REST API Service
 
-A FastAPI-based REST API service.
+A RESTful API service built with FastAPI and SQLite.
+
+## Features
+
+- FastAPI framework for efficient API development
+- SQLite database for data storage
+- Alembic for database migrations
+- Pydantic for data validation
+- SQLAlchemy ORM for database interactions
+- OpenAPI documentation (Swagger UI and ReDoc)
+- Health endpoint for application monitoring
 
 ## Project Structure
 
 ```
-/projects/restapiservice-12nbts/
+.
 ├── alembic.ini                  # Alembic configuration
-├── app                          # Application package
-│   ├── api                      # API endpoints
-│   │   └── endpoints            # Route handlers
-│   ├── core                     # Core modules
+├── app/                         # Application package
+│   ├── api/                     # API endpoints
+│   │   ├── endpoints/           # Route handlers
+│   │   └── api.py               # API router
+│   ├── core/                    # Core modules
 │   │   └── config.py            # Application settings
-│   ├── db                       # Database modules
+│   ├── db/                      # Database modules
 │   │   ├── base.py              # Import all models for Alembic
 │   │   ├── base_class.py        # SQLAlchemy Base class
 │   │   └── session.py           # Database session management
-│   ├── models                   # SQLAlchemy models
-│   ├── schemas                  # Pydantic schemas for request/response
-│   └── services                 # Business logic
+│   ├── models/                  # SQLAlchemy models
+│   ├── schemas/                 # Pydantic schemas
+│   └── services/                # Business logic
 ├── main.py                      # Application entry point
-├── migrations                   # Alembic migrations
-│   └── versions                 # Migration scripts
+├── migrations/                  # Alembic migrations
+│   └── versions/                # Migration scripts
 ├── requirements.txt             # Project dependencies
-└── tests                        # Test modules
+└── tests/                       # Test modules
 ```
 
-## Getting Started
+## API Endpoints
+
+- `/api/v1/users` - User CRUD operations
+- `/api/v1/items` - Item CRUD operations
+- `/health` - Health check endpoint
+- `/docs` - Swagger UI documentation
+- `/redoc` - ReDoc documentation
+
+## Installation
+
+1. Clone the repository:
+```bash
+git clone https://github.com/yourusername/restapiservice.git
+cd restapiservice
+```
 
-1. Clone the repository
 2. Install dependencies:
-   ```
-   pip install -r requirements.txt
-   ```
-3. Run the application:
-   ```
-   uvicorn main:app --reload
-   ```
+```bash
+pip install -r requirements.txt
+```
+
+3. Run database migrations:
+```bash
+alembic upgrade head
+```
+
+4. Start the server:
+```bash
+uvicorn main:app --reload
+```
+
+The API will be available at http://localhost:8000.
+
+## Development
+
+### Database Migrations
+
+To create a new migration after modifying models:
+
+```bash
+alembic revision --autogenerate -m "Description of changes"
+```
+
+To apply migrations:
+
+```bash
+alembic upgrade head
+```
+
+### Linting
+
+The project uses Ruff for linting and formatting:
+
+```bash
+ruff check .
+ruff format .
+```
 
 ## API Documentation
 
-- Swagger UI: [http://localhost:8000/docs](http://localhost:8000/docs)
-- ReDoc: [http://localhost:8000/redoc](http://localhost:8000/redoc)
+Once the server is running, you can access:
 
-## Health Check
+- Swagger UI: http://localhost:8000/docs
+- ReDoc: http://localhost:8000/redoc
 
-- GET `/health`: Returns a health status of the application
+## License
+
+This project is licensed under the MIT License.

alembic.ini

@@ -35,6 +35,7 @@ script_location = migrations
 # are written from script.py.mako
 # output_encoding = utf-8
 
+# SQLite URL example
 sqlalchemy.url = sqlite:////app/storage/db/db.sqlite
 
 [post_write_hooks]
@@ -81,4 +82,4 @@ formatter = generic
 
 [formatter_generic]
 format = %(levelname)-5.5s [%(name)s] %(message)s
-datefmt = %H:%M:%S
+datefmt = %H:%M:%S

app/api/api.py

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

app/api/endpoints/health.py

@@ -0,0 +1,17 @@
+from fastapi import APIRouter
+
+from app.core.config import settings
+from app.schemas.response import HealthResponse
+
+router = APIRouter()
+
+
+@router.get("", response_model=HealthResponse)
+async def health_check() -> HealthResponse:
+    """
+    Health check endpoint to verify the API is working.
+    """
+    return HealthResponse(
+        status="ok",
+        version=settings.VERSION,
+    )

app/api/endpoints/items.py

@@ -0,0 +1,103 @@
+from typing import Any, List
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from app.db.session import get_db
+from app.schemas.item import ItemCreate, Item as ItemSchema, ItemUpdate
+from app.schemas.response import DataResponse
+from app.services.item import (
+    create_item,
+    get_item,
+    get_items,
+    update_item,
+    delete_item,
+)
+
+router = APIRouter()
+
+
+@router.post(
+    "/", response_model=DataResponse[ItemSchema], status_code=status.HTTP_201_CREATED
+)
+def create_new_item(
+    *,
+    item_in: ItemCreate,
+    owner_id: int,
+    db: Session = Depends(get_db),
+) -> Any:
+    """
+    Create new item.
+    """
+    item = create_item(db, obj_in=item_in, owner_id=owner_id)
+    return DataResponse(data=item, message="Item created successfully")
+
+
+@router.get("/", response_model=DataResponse[List[ItemSchema]])
+def read_items(
+    skip: int = 0,
+    limit: int = 100,
+    db: Session = Depends(get_db),
+) -> Any:
+    """
+    Retrieve items.
+    """
+    items = get_items(db, skip=skip, limit=limit)
+    return DataResponse(data=items, message="Items retrieved successfully")
+
+
+@router.get("/{item_id}", response_model=DataResponse[ItemSchema])
+def read_item(
+    item_id: int,
+    db: Session = Depends(get_db),
+) -> Any:
+    """
+    Get item by ID.
+    """
+    item = get_item(db, id=item_id)
+    if not item:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Item not found",
+        )
+    return DataResponse(data=item, message="Item retrieved successfully")
+
+
+@router.put("/{item_id}", response_model=DataResponse[ItemSchema])
+def update_item_info(
+    *,
+    item_id: int,
+    item_in: ItemUpdate,
+    db: Session = Depends(get_db),
+) -> Any:
+    """
+    Update item.
+    """
+    item = get_item(db, id=item_id)
+    if not item:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Item not found",
+        )
+    item = update_item(db, db_obj=item, obj_in=item_in)
+    return DataResponse(data=item, message="Item updated successfully")
+
+
+@router.delete(
+    "/{item_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None
+)
+def delete_item_by_id(
+    item_id: int,
+    db: Session = Depends(get_db),
+) -> None:
+    """
+    Delete item.
+    """
+    item = get_item(db, id=item_id)
+    if not item:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="Item not found",
+        )
+    delete_item(db, id=item_id)
+    return None

app/api/endpoints/users.py

@@ -0,0 +1,109 @@
+from typing import Any, List
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from app.db.session import get_db
+from app.schemas.user import UserCreate, User as UserSchema, UserUpdate
+from app.schemas.response import DataResponse
+from app.services.user import (
+    create_user,
+    get_user,
+    get_user_by_email,
+    get_users,
+    update_user,
+    delete_user,
+)
+
+router = APIRouter()
+
+
+@router.post(
+    "/", response_model=DataResponse[UserSchema], status_code=status.HTTP_201_CREATED
+)
+def create_new_user(
+    *,
+    user_in: UserCreate,
+    db: Session = Depends(get_db),
+) -> Any:
+    """
+    Create new user.
+    """
+    user = get_user_by_email(db, email=user_in.email)
+    if user:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="The user with this email already exists.",
+        )
+    user = create_user(db, obj_in=user_in)
+    return DataResponse(data=user, message="User created successfully")
+
+
+@router.get("/", response_model=DataResponse[List[UserSchema]])
+def read_users(
+    skip: int = 0,
+    limit: int = 100,
+    db: Session = Depends(get_db),
+) -> Any:
+    """
+    Retrieve users.
+    """
+    users = get_users(db, skip=skip, limit=limit)
+    return DataResponse(data=users, message="Users retrieved successfully")
+
+
+@router.get("/{user_id}", response_model=DataResponse[UserSchema])
+def read_user(
+    user_id: int,
+    db: Session = Depends(get_db),
+) -> Any:
+    """
+    Get user by ID.
+    """
+    user = get_user(db, id=user_id)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="User not found",
+        )
+    return DataResponse(data=user, message="User retrieved successfully")
+
+
+@router.put("/{user_id}", response_model=DataResponse[UserSchema])
+def update_user_info(
+    *,
+    user_id: int,
+    user_in: UserUpdate,
+    db: Session = Depends(get_db),
+) -> Any:
+    """
+    Update user.
+    """
+    user = get_user(db, id=user_id)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="User not found",
+        )
+    user = update_user(db, db_obj=user, obj_in=user_in)
+    return DataResponse(data=user, message="User updated successfully")
+
+
+@router.delete(
+    "/{user_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None
+)
+def delete_user_by_id(
+    user_id: int,
+    db: Session = Depends(get_db),
+) -> None:
+    """
+    Delete user.
+    """
+    user = get_user(db, id=user_id)
+    if not user:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="User not found",
+        )
+    delete_user(db, id=user_id)
+    return None

app/core/config.py

@@ -1,21 +1,22 @@
-import secrets
-from pathlib import Path
-from pydantic import BaseSettings
+from pydantic_settings import BaseSettings
+
 
 class Settings(BaseSettings):
-    PROJECT_NAME: str = "REST API Service"
+    """
+    Application settings.
+    """
+
     API_V1_STR: str = "/api/v1"
+    PROJECT_NAME: str = "REST API Service"
+    PROJECT_DESCRIPTION: str = "A REST API service built with FastAPI and SQLite"
     VERSION: str = "0.1.0"
-    SECRET_KEY: str = secrets.token_urlsafe(32)
-    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8  # 8 days
-    
-    # Database
-    DB_DIR = Path("/app") / "storage" / "db"
-    DB_DIR.mkdir(parents=True, exist_ok=True)
-    SQLALCHEMY_DATABASE_URL: str = f"sqlite:///{DB_DIR}/db.sqlite"
+
+    # CORS settings
+    BACKEND_CORS_ORIGINS: list[str] = ["*"]
 
     class Config:
         env_file = ".env"
         case_sensitive = True
 
+
 settings = Settings()

app/db/base.py

@@ -1,3 +1,4 @@
-# Import all models here for Alembic to detect
-from app.db.base_class import Base
-# Example: from app.models.item import Item
+# Import all the models here for Alembic to detect
+from app.db.base_class import Base  # noqa
+from app.models.item import Item  # noqa
+from app.models.user import User  # noqa

app/db/base_class.py

@@ -1,11 +1,14 @@
 from typing import Any
+
 from sqlalchemy.ext.declarative import as_declarative, declared_attr
 
+
 @as_declarative()
 class Base:
     id: Any
     __name__: str
-    
+
+    # Generate __tablename__ automatically based on class name
     @declared_attr
     def __tablename__(cls) -> str:
         return cls.__name__.lower()

app/db/session.py

@@ -1,17 +1,26 @@
+from pathlib import Path
+
 from sqlalchemy import create_engine
-from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import sessionmaker
 
-from app.core.config import settings
+# Create the directory for the database if it doesn't exist
+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(
-    settings.SQLALCHEMY_DATABASE_URL,
-    connect_args={"check_same_thread": False}  # Only needed for SQLite
+    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
 )
+
 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
-Base = declarative_base()
+
 
 def get_db():
+    """
+    Dependency function to get a database session.
+    Yields a SQLAlchemy session and ensures it's closed after use.
+    """
     db = SessionLocal()
     try:
         yield db

app/models/__init__.py

@@ -0,0 +1,2 @@
+from app.models.item import Item  # noqa
+from app.models.user import User  # noqa

app/models/item.py

@@ -0,0 +1,18 @@
+from sqlalchemy import Column, ForeignKey, Integer, String, Text
+from sqlalchemy.orm import relationship
+
+from app.db.base_class import Base
+
+
+class Item(Base):
+    """
+    Item model for storing item information.
+    """
+
+    id = Column(Integer, primary_key=True, index=True)
+    title = Column(String(100), index=True, nullable=False)
+    description = Column(Text)
+    owner_id = Column(Integer, ForeignKey("user.id"))
+
+    # Define relationship with User
+    owner = relationship("User", back_populates="items")

app/models/user.py

@@ -0,0 +1,20 @@
+from sqlalchemy import Boolean, Column, Integer, String
+from sqlalchemy.orm import relationship
+
+from app.db.base_class import Base
+
+
+class User(Base):
+    """
+    User model for storing user information.
+    """
+
+    id = Column(Integer, primary_key=True, index=True)
+    email = Column(String, unique=True, index=True, nullable=False)
+    username = Column(String, unique=True, index=True, nullable=False)
+    hashed_password = Column(String, nullable=False)
+    is_active = Column(Boolean, default=True)
+    is_superuser = Column(Boolean, default=False)
+
+    # Define relationship with Items
+    items = relationship("Item", back_populates="owner")

app/schemas/__init__.py

@@ -0,0 +1,9 @@
+from app.schemas.base import BaseSchema  # noqa
+from app.schemas.item import Item, ItemCreate, ItemInDB, ItemUpdate  # noqa
+from app.schemas.response import (
+    DataResponse,
+    ErrorResponse,
+    HealthResponse,
+    ResponseBase,
+)  # noqa
+from app.schemas.user import User, UserCreate, UserInDB, UserUpdate  # noqa

app/schemas/base.py

@@ -0,0 +1,9 @@
+from pydantic import BaseModel, ConfigDict
+
+
+class BaseSchema(BaseModel):
+    """
+    Base class for all schemas.
+    """
+
+    model_config = ConfigDict(from_attributes=True)

app/schemas/item.py

@@ -0,0 +1,55 @@
+from typing import Optional
+
+
+from app.schemas.base import BaseSchema
+
+
+class ItemBase(BaseSchema):
+    """
+    Base schema for an item.
+    """
+
+    title: Optional[str] = None
+    description: Optional[str] = None
+
+
+class ItemCreate(ItemBase):
+    """
+    Schema for creating a new item.
+    """
+
+    title: str
+
+
+class ItemUpdate(ItemBase):
+    """
+    Schema for updating an item.
+    """
+
+    pass
+
+
+class ItemInDBBase(ItemBase):
+    """
+    Schema for an item in the database.
+    """
+
+    id: int
+    title: str
+    owner_id: int
+
+
+class Item(ItemInDBBase):
+    """
+    Schema for returning an item.
+    """
+
+    pass
+
+
+class ItemInDB(ItemInDBBase):
+    """
+    Schema for an item in the database.
+    """
+
+    pass

app/schemas/response.py

@@ -0,0 +1,41 @@
+from typing import Any, Dict, Generic, List, Optional, TypeVar, Union
+
+from pydantic import BaseModel
+
+T = TypeVar("T")
+
+
+class ResponseBase(BaseModel):
+    """
+    Base response schema.
+    """
+
+    success: bool = True
+    message: str = "Operation successful"
+
+
+class DataResponse(ResponseBase, Generic[T]):
+    """
+    Response schema with data.
+    """
+
+    data: T
+
+
+class ErrorResponse(ResponseBase):
+    """
+    Error response schema.
+    """
+
+    success: bool = False
+    message: str = "An error occurred"
+    error_details: Optional[Union[List[Dict[str, Any]], Dict[str, Any], str]] = None
+
+
+class HealthResponse(BaseModel):
+    """
+    Health check response schema.
+    """
+
+    status: str = "ok"
+    version: str

app/schemas/user.py

@@ -0,0 +1,60 @@
+from typing import Optional
+
+from pydantic import EmailStr
+
+from app.schemas.base import BaseSchema
+
+
+class UserBase(BaseSchema):
+    """
+    Base schema for a user.
+    """
+
+    email: Optional[EmailStr] = None
+    username: Optional[str] = None
+    is_active: Optional[bool] = True
+    is_superuser: bool = False
+
+
+class UserCreate(UserBase):
+    """
+    Schema for creating a new user.
+    """
+
+    email: EmailStr
+    username: str
+    password: str
+
+
+class UserUpdate(UserBase):
+    """
+    Schema for updating a user.
+    """
+
+    password: Optional[str] = None
+
+
+class UserInDBBase(UserBase):
+    """
+    Schema for a user in the database.
+    """
+
+    id: int
+    email: EmailStr
+    username: str
+
+
+class User(UserInDBBase):
+    """
+    Schema for returning a user.
+    """
+
+    pass
+
+
+class UserInDB(UserInDBBase):
+    """
+    Schema for a user in the database with the hashed password.
+    """
+
+    hashed_password: str

app/services/item.py

@@ -0,0 +1,71 @@
+from typing import List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.models.item import Item
+from app.schemas.item import ItemCreate, ItemUpdate
+
+
+def get_item(db: Session, id: int) -> Optional[Item]:
+    """
+    Get an item by ID.
+    """
+    return db.query(Item).filter(Item.id == id).first()
+
+
+def get_items(db: Session, skip: int = 0, limit: int = 100) -> List[Item]:
+    """
+    Get multiple items.
+    """
+    return db.query(Item).offset(skip).limit(limit).all()
+
+
+def get_user_items(
+    db: Session, owner_id: int, skip: int = 0, limit: int = 100
+) -> List[Item]:
+    """
+    Get items for a specific user.
+    """
+    return (
+        db.query(Item).filter(Item.owner_id == owner_id).offset(skip).limit(limit).all()
+    )
+
+
+def create_item(db: Session, obj_in: ItemCreate, owner_id: int) -> Item:
+    """
+    Create a new item.
+    """
+    db_obj = Item(
+        title=obj_in.title,
+        description=obj_in.description,
+        owner_id=owner_id,
+    )
+    db.add(db_obj)
+    db.commit()
+    db.refresh(db_obj)
+    return db_obj
+
+
+def update_item(db: Session, db_obj: Item, obj_in: ItemUpdate) -> Item:
+    """
+    Update an item.
+    """
+    update_data = obj_in.dict(exclude_unset=True)
+
+    for field, value in update_data.items():
+        setattr(db_obj, field, value)
+
+    db.add(db_obj)
+    db.commit()
+    db.refresh(db_obj)
+    return db_obj
+
+
+def delete_item(db: Session, id: int) -> None:
+    """
+    Delete an item.
+    """
+    item = db.query(Item).get(id)
+    if item:
+        db.delete(item)
+        db.commit()

app/services/user.py

@@ -0,0 +1,75 @@
+from typing import List, Optional
+
+from sqlalchemy.orm import Session
+
+from app.models.user import User
+from app.schemas.user import UserCreate, UserUpdate
+
+
+def get_user(db: Session, id: int) -> Optional[User]:
+    """
+    Get a user by ID.
+    """
+    return db.query(User).filter(User.id == id).first()
+
+
+def get_user_by_email(db: Session, email: str) -> Optional[User]:
+    """
+    Get a user by email.
+    """
+    return db.query(User).filter(User.email == email).first()
+
+
+def get_users(db: Session, skip: int = 0, limit: int = 100) -> List[User]:
+    """
+    Get multiple users.
+    """
+    return db.query(User).offset(skip).limit(limit).all()
+
+
+def create_user(db: Session, obj_in: UserCreate) -> User:
+    """
+    Create a new user.
+    """
+    # In a real application, you'd hash the password
+    db_obj = User(
+        email=obj_in.email,
+        username=obj_in.username,
+        hashed_password=obj_in.password,  # This should be hashed in a real app
+        is_active=obj_in.is_active,
+        is_superuser=obj_in.is_superuser,
+    )
+    db.add(db_obj)
+    db.commit()
+    db.refresh(db_obj)
+    return db_obj
+
+
+def update_user(db: Session, db_obj: User, obj_in: UserUpdate) -> User:
+    """
+    Update a user.
+    """
+    update_data = obj_in.dict(exclude_unset=True)
+
+    # Handle password updates
+    if "password" in update_data and update_data["password"]:
+        # In a real application, you'd hash the password
+        update_data["hashed_password"] = update_data.pop("password")
+
+    for field, value in update_data.items():
+        setattr(db_obj, field, value)
+
+    db.add(db_obj)
+    db.commit()
+    db.refresh(db_obj)
+    return db_obj
+
+
+def delete_user(db: Session, id: int) -> None:
+    """
+    Delete a user.
+    """
+    user = db.query(User).get(id)
+    if user:
+        db.delete(user)
+        db.commit()

main.py

@@ -1,15 +1,51 @@
 import uvicorn
 from fastapi import FastAPI
-from app.api.endpoints import router as api_router
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import RedirectResponse
+
+from app.api.api import api_router
 from app.core.config import settings
 
-app = FastAPI(title=settings.PROJECT_NAME, version=settings.VERSION)
+app = FastAPI(
+    title=settings.PROJECT_NAME,
+    description=settings.PROJECT_DESCRIPTION,
+    version=settings.VERSION,
+    openapi_url=f"{settings.API_V1_STR}/openapi.json",
+    docs_url="/docs",
+    redoc_url="/redoc",
+)
 
+# Set up CORS middleware
+if settings.BACKEND_CORS_ORIGINS:
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+# Include API router
 app.include_router(api_router, prefix=settings.API_V1_STR)
 
-@app.get('/health', tags=['Health'])
-async def health_check():
-    return {"status": "healthy"}
+
+# Root endpoint
+@app.get("/", include_in_schema=False)
+def root() -> RedirectResponse:
+    """
+    Redirect to API documentation.
+    """
+    return RedirectResponse(url="/docs")
+
+
+# Health check endpoint directly in main.py
+@app.get("/health", tags=["health"])
+async def root_health_check():
+    """
+    Root health check endpoint.
+    """
+    return {"status": "ok", "version": settings.VERSION}
+
 
 if __name__ == "__main__":
     uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

migrations/env.py

@@ -1,9 +1,7 @@
 from logging.config import fileConfig
 
-from sqlalchemy import engine_from_config
-from sqlalchemy import pool
-
 from alembic import context
+from sqlalchemy import engine_from_config, pool
 
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.
@@ -15,7 +13,10 @@ fileConfig(config.config_file_name)
 
 # add your model's MetaData object here
 # for 'autogenerate' support
+# from myapp import mymodel
+# target_metadata = mymodel.Base.metadata
 from app.db.base import Base  # noqa
+
 target_metadata = Base.metadata
 
 # other values from the config, defined by the needs of env.py,
@@ -25,8 +26,7 @@ target_metadata = Base.metadata
 
 
 def run_migrations_offline():
-    """
-    Run migrations in 'offline' mode.
+    """Run migrations in 'offline' mode.
 
     This configures the context with just a URL
     and not an Engine, though an Engine is acceptable
@@ -50,8 +50,7 @@ def run_migrations_offline():
 
 
 def run_migrations_online():
-    """
-    Run migrations in 'online' mode.
+    """Run migrations in 'online' mode.
 
     In this scenario we need to create an Engine
     and associate a connection with the context.
@@ -64,8 +63,11 @@ def run_migrations_online():
     )
 
     with connectable.connect() as connection:
+        is_sqlite = connection.dialect.name == "sqlite"
         context.configure(
-            connection=connection, target_metadata=target_metadata
+            connection=connection,
+            target_metadata=target_metadata,
+            render_as_batch=is_sqlite,  # Key configuration for SQLite
         )
 
         with context.begin_transaction():

migrations/script.py.mako

@@ -21,4 +21,4 @@ def upgrade():
 
 
 def downgrade():
-    ${downgrades if downgrades else "pass"}
+    ${downgrades if downgrades else "pass"}

migrations/versions/001_initial_tables.py

@@ -0,0 +1,63 @@
+"""Initial tables
+
+Revision ID: 001
+Revises:
+Create Date: 2023-09-14
+
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = "001"
+down_revision = None
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # Create user table
+    op.create_table(
+        "user",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("email", sa.String(), nullable=False),
+        sa.Column("username", sa.String(), nullable=False),
+        sa.Column("hashed_password", sa.String(), nullable=False),
+        sa.Column("is_active", sa.Boolean(), nullable=True),
+        sa.Column("is_superuser", sa.Boolean(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(op.f("ix_user_email"), "user", ["email"], unique=True)
+    op.create_index(op.f("ix_user_id"), "user", ["id"], unique=False)
+    op.create_index(op.f("ix_user_username"), "user", ["username"], unique=True)
+
+    # Create item table
+    op.create_table(
+        "item",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("title", sa.String(length=100), nullable=False),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("owner_id", sa.Integer(), nullable=True),
+        sa.ForeignKeyConstraint(
+            ["owner_id"],
+            ["user.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(op.f("ix_item_id"), "item", ["id"], unique=False)
+    op.create_index(op.f("ix_item_title"), "item", ["title"], unique=False)
+
+
+def downgrade():
+    # Drop item table
+    op.drop_index(op.f("ix_item_title"), table_name="item")
+    op.drop_index(op.f("ix_item_id"), table_name="item")
+    op.drop_table("item")
+
+    # Drop user table
+    op.drop_index(op.f("ix_user_username"), table_name="user")
+    op.drop_index(op.f("ix_user_id"), table_name="user")
+    op.drop_index(op.f("ix_user_email"), table_name="user")
+    op.drop_table("user")

requirements.txt

@@ -1,11 +1,10 @@
-fastapi>=0.68.0
-pydantic>=1.8.2
-uvicorn>=0.15.0
-sqlalchemy>=1.4.23
-alembic>=1.7.1
-python-dotenv>=0.19.0
-python-jose>=3.3.0
-passlib>=1.7.4
-python-multipart>=0.0.5
-email-validator>=1.1.3
-ruff>=0.0.290
+fastapi==0.103.1
+uvicorn==0.23.2
+sqlalchemy==2.0.20
+alembic==1.12.0
+pydantic==2.3.0
+pydantic-settings==2.0.3
+python-dotenv==1.0.0
+pytest==7.4.2
+httpx==0.24.1
+ruff==0.0.291