Complete Todo App implementation

- Added filtering and pagination for todo listings
- Fixed Alembic migration setup
- Enhanced CRUD operations
- Updated documentation with comprehensive README
- Linted code with Ruff
This commit is contained in:
Automated Action 2025-06-17 02:13:03 +00:00
parent 790a9f178e
commit 1fca93299a
36 changed files with 1146 additions and 2 deletions

147
README.md
View File

@ -1,3 +1,146 @@
# FastAPI Application
# TodoApp API - FastAPI Backend
This is a FastAPI application bootstrapped by BackendIM, the AI-powered backend generation platform.
A robust, RESTful API for managing todos, built with FastAPI and SQLite.
## Features
- 🔐 JWT Authentication
- 📝 Todo CRUD operations
- 👤 User management
- 🔍 Advanced todo filtering and pagination
- 📄 API documentation (via Swagger UI and ReDoc)
- 🔄 Database migrations (Alembic)
## Requirements
- Python 3.8+
- FastAPI
- SQLAlchemy
- Alembic
- Pydantic
- SQLite
## Installation
1. Clone the repository
2. Install dependencies:
```bash
pip install -r requirements.txt
```
3. Run database migrations:
```bash
alembic upgrade head
```
## Running the Application
Start the server with:
```bash
uvicorn main:app --reload
```
The API will be available at http://localhost:8000.
- API Documentation: http://localhost:8000/docs
- Alternative Documentation: http://localhost:8000/redoc
- OpenAPI Schema: http://localhost:8000/openapi.json
- Health Check: http://localhost:8000/health
## Environment Variables
The application can be configured using the following environment variables:
| Variable | Description | Default |
|----------|-------------|---------|
| SECRET_KEY | Secret key for JWT encoding | Auto-generated |
| ACCESS_TOKEN_EXPIRE_MINUTES | JWT token expiration time (minutes) | 11520 (8 days) |
## API Endpoints
### Authentication
- `POST /api/v1/auth/register` - Register a new user
- `POST /api/v1/auth/login` - Login and get access token
### Users
- `GET /api/v1/users/` - List all users
- `GET /api/v1/users/me` - Get current user details
- `PUT /api/v1/users/me` - Update current user
- `GET /api/v1/users/{user_id}` - Get user by ID
### Todos
- `GET /api/v1/todos/` - List todos (with filtering and pagination)
- `POST /api/v1/todos/` - Create a new todo
- `GET /api/v1/todos/{id}` - Get todo by ID
- `PUT /api/v1/todos/{id}` - Update a todo
- `DELETE /api/v1/todos/{id}` - Delete a todo
#### Todo Filtering
The `GET /api/v1/todos/` endpoint supports the following query parameters:
- `skip`: Number of records to skip (default: 0)
- `limit`: Maximum number of records to return (default: 100)
- `title`: Filter by title (contains search)
- `is_completed`: Filter by completion status (true/false)
## Database Schema
### User Model
```
id: Integer (Primary Key)
email: String (Unique, Indexed)
hashed_password: String
is_active: Boolean (Default: True)
```
### Todo Model
```
id: Integer (Primary Key)
title: String (Indexed)
description: Text (Optional)
is_completed: Boolean (Default: False)
owner_id: Integer (Foreign Key to User)
```
## Development
### Code Structure
- `app/`: Main application package
- `api/`: API routes and dependencies
- `core/`: Core functionality (config, security)
- `crud/`: CRUD operations
- `db/`: Database setup and session management
- `models/`: SQLAlchemy models
- `schemas/`: Pydantic schemas
- `storage/`: Storage for database and other files
- `migrations/`: Alembic migrations
- `main.py`: Application entry point
### Adding New Models
1. Create a new model in `app/models/`
2. Import the model in `app/db/base_class.py`
3. Create corresponding Pydantic schemas in `app/schemas/`
4. Create CRUD operations in `app/crud/`
5. Create API endpoints in `app/api/v1/endpoints/`
6. Generate a new migration:
```bash
alembic revision -m "description"
```
7. Edit the migration file manually
8. Apply the migration:
```bash
alembic upgrade head
```
## License
MIT License

85
alembic.ini Normal file
View File

@ -0,0 +1,85 @@
# A generic, single database configuration.
[alembic]
# path to migration scripts
script_location = migrations
# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s
# timezone to use when rendering the date
# within the migration file as well as the filename.
# string value is passed to dateutil.tz.gettz()
# leave blank for localtime
# timezone =
# max length of characters to apply to the
# "slug" field
# truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version location specification; this defaults
# to migrations/versions. When using multiple version
# directories, initial revisions must be specified with --version-path
# version_locations = %(here)s/bar %(here)s/bat migrations/versions
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
sqlalchemy.url = sqlite:////app/storage/db/db.sqlite
[post_write_hooks]
# post_write_hooks defines scripts or Python functions that are run
# on newly generated revision scripts. See the documentation for further
# detail and examples
# format using "black" - use the console_scripts runner, against the "black" entrypoint
# hooks=black
# black.type=console_scripts
# black.entrypoint=black
# black.options=-l 79
# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic
[handlers]
keys = console
[formatters]
keys = generic
[logger_root]
level = WARN
handlers = console
qualname =
[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine
[logger_alembic]
level = INFO
handlers =
qualname = alembic
[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic
[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

0
app/__init__.py Normal file
View File

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

52
app/api/deps.py Normal file
View File

@ -0,0 +1,52 @@
from typing import Generator
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt
from pydantic import ValidationError
from sqlalchemy.orm import Session
from app.core.config import settings
from app.crud.crud_user import get_user
from app.db.session import SessionLocal
from app.models.user import User
from app.schemas.token import TokenPayload
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl=f"{settings.API_V1_STR}/auth/login"
)
def get_db() -> Generator:
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_current_user(
db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> User:
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=["HS256"]
)
token_data = TokenPayload(**payload)
except (jwt.JWTError, ValidationError):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Could not validate credentials",
)
user = get_user(db, user_id=token_data.sub)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
def get_current_active_user(
current_user: User = Depends(get_current_user),
) -> User:
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user

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

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

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

View File

View File

@ -0,0 +1,68 @@
from datetime import timedelta
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from app.api.deps import get_db
from app.core.config import settings
from app.core.security import create_access_token
from app.crud.crud_user import authenticate, create_user, get_user_by_email
from app.schemas.token import Token
from app.schemas.user import UserCreate
router = APIRouter()
@router.post("/login", response_model=Token)
def login_access_token(
db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()
) -> Any:
"""
OAuth2 compatible token login, get an access token for future requests
"""
user = authenticate(
db, email=form_data.username, password=form_data.password
)
if not user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Incorrect email or password",
)
elif not user.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST, detail="Inactive user"
)
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return {
"access_token": create_access_token(
user.id, expires_delta=access_token_expires
),
"token_type": "bearer",
}
@router.post("/register", response_model=Token)
def register_user(
*,
db: Session = Depends(get_db),
user_in: UserCreate,
) -> Any:
"""
Register a new user and return an access token
"""
user = get_user_by_email(db, email=user_in.email)
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="A user with this email already exists",
)
user = create_user(db, obj_in=user_in)
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return {
"access_token": create_access_token(
user.id, expires_delta=access_token_expires
),
"token_type": "bearer",
}

View File

@ -0,0 +1,116 @@
from typing import Any, List, Optional
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.api.deps import get_current_active_user, get_db
from app.crud.crud_todo import (
create_todo,
delete_todo,
get_todo,
get_todos,
update_todo,
)
from app.models.user import User
from app.schemas.todo import Todo, TodoCreate, TodoUpdate
router = APIRouter()
@router.get("/", response_model=List[Todo])
def read_todos(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
title: Optional[str] = None,
is_completed: Optional[bool] = None,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve todos with optional filtering.
- **skip**: Number of records to skip for pagination
- **limit**: Maximum number of records to return
- **title**: Filter by title (contains search)
- **is_completed**: Filter by completion status
"""
todos = get_todos(
db=db,
owner_id=current_user.id,
skip=skip,
limit=limit,
title=title,
is_completed=is_completed
)
return todos
@router.post("/", response_model=Todo)
def create_todo_item(
*,
db: Session = Depends(get_db),
todo_in: TodoCreate,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Create new todo.
"""
todo = create_todo(db=db, todo_in=todo_in, owner_id=current_user.id)
return todo
@router.get("/{id}", response_model=Todo)
def read_todo(
*,
db: Session = Depends(get_db),
id: int,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Get todo by ID.
"""
todo = get_todo(db=db, todo_id=id)
if not todo:
raise HTTPException(status_code=404, detail="Todo not found")
if todo.owner_id != current_user.id:
raise HTTPException(status_code=400, detail="Not enough permissions")
return todo
@router.put("/{id}", response_model=Todo)
def update_todo_item(
*,
db: Session = Depends(get_db),
id: int,
todo_in: TodoUpdate,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Update a todo.
"""
todo = get_todo(db=db, todo_id=id)
if not todo:
raise HTTPException(status_code=404, detail="Todo not found")
if todo.owner_id != current_user.id:
raise HTTPException(status_code=400, detail="Not enough permissions")
todo = update_todo(db=db, db_obj=todo, obj_in=todo_in)
return todo
@router.delete("/{id}", response_model=None, status_code=status.HTTP_204_NO_CONTENT)
def delete_todo_item(
*,
db: Session = Depends(get_db),
id: int,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Delete a todo.
"""
todo = get_todo(db=db, todo_id=id)
if not todo:
raise HTTPException(status_code=404, detail="Todo not found")
if todo.owner_id != current_user.id:
raise HTTPException(status_code=400, detail="Not enough permissions")
delete_todo(db=db, todo_id=id)
return None

View File

@ -0,0 +1,70 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.api.deps import get_current_active_user, get_db
from app.crud.crud_user import get_user, get_users, update_user
from app.models.user import User
from app.schemas.user import User as UserSchema
from app.schemas.user import UserUpdate
router = APIRouter()
@router.get("/", response_model=List[UserSchema])
def read_users(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve users.
"""
users = get_users(db, skip=skip, limit=limit)
return users
@router.get("/me", response_model=UserSchema)
def read_user_me(
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Get current user.
"""
return current_user
@router.put("/me", response_model=UserSchema)
def update_user_me(
*,
db: Session = Depends(get_db),
user_in: UserUpdate,
current_user: User = Depends(get_current_active_user),
) -> Any:
"""
Update own user.
"""
user = update_user(db, db_obj=current_user, obj_in=user_in)
return user
@router.get("/{user_id}", response_model=UserSchema)
def read_user_by_id(
user_id: int,
current_user: User = Depends(get_current_active_user),
db: Session = Depends(get_db),
) -> Any:
"""
Get a specific user by id.
"""
user = get_user(db, user_id=user_id)
if user == current_user:
return user
if not user:
raise HTTPException(
status_code=404,
detail="The user with this id does not exist in the system",
)
return user

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

36
app/core/config.py Normal file
View File

@ -0,0 +1,36 @@
import secrets
from typing import List
from pydantic import field_validator
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
API_V1_STR: str = "/api/v1"
SECRET_KEY: str = secrets.token_urlsafe(32)
# 60 minutes * 24 hours * 8 days = 8 days
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8
# SERVER_NAME: str
# SERVER_HOST: AnyHttpUrl
# BACKEND_CORS_ORIGINS is a JSON-formatted list of origins
# e.g: '["http://localhost", "http://localhost:4200", "http://localhost:3000", \
# "http://localhost:8080", "http://local.dockertoolbox.tiangolo.com"]'
BACKEND_CORS_ORIGINS: List[str] = ["*"]
@field_validator("BACKEND_CORS_ORIGINS", mode="before")
@classmethod
def assemble_cors_origins(cls, v: str | List[str]) -> List[str] | str:
if isinstance(v, str) and not v.startswith("["):
return [i.strip() for i in v.split(",")]
elif isinstance(v, (list, str)):
return v
raise ValueError(v)
PROJECT_NAME: str = "TodoApp API"
class Config:
case_sensitive = True
env_file = ".env"
settings = Settings()

31
app/core/security.py Normal file
View File

@ -0,0 +1,31 @@
from datetime import datetime, timedelta
from typing import Any, Union
from jose import jwt
from passlib.context import CryptContext
from app.core.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def create_access_token(
subject: Union[str, Any], expires_delta: timedelta = None
) -> str:
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
)
to_encode = {"exp": expire, "sub": str(subject)}
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")
return encoded_jwt
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)

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

61
app/crud/crud_todo.py Normal file
View File

@ -0,0 +1,61 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.models.todo import Todo
from app.schemas.todo import TodoCreate, TodoUpdate
def get_todo(db: Session, todo_id: int) -> Optional[Todo]:
return db.query(Todo).filter(Todo.id == todo_id).first()
def get_todos(
db: Session,
owner_id: int,
skip: int = 0,
limit: int = 100,
title: Optional[str] = None,
is_completed: Optional[bool] = None
) -> List[Todo]:
query = db.query(Todo).filter(Todo.owner_id == owner_id)
if title:
query = query.filter(Todo.title.contains(title))
if is_completed is not None:
query = query.filter(Todo.is_completed == is_completed)
return query.offset(skip).limit(limit).all()
def create_todo(db: Session, todo_in: TodoCreate, owner_id: int) -> Todo:
db_todo = Todo(
title=todo_in.title,
description=todo_in.description,
is_completed=todo_in.is_completed,
owner_id=owner_id
)
db.add(db_todo)
db.commit()
db.refresh(db_todo)
return db_todo
def update_todo(
db: Session, db_obj: Todo, obj_in: TodoUpdate
) -> Todo:
update_data = obj_in.model_dump(exclude_unset=True)
for field in update_data:
setattr(db_obj, field, update_data[field])
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def delete_todo(db: Session, todo_id: int) -> Todo:
todo = db.query(Todo).filter(Todo.id == todo_id).first()
db.delete(todo)
db.commit()
return todo

61
app/crud/crud_user.py Normal file
View File

@ -0,0 +1,61 @@
from typing import Any, Dict, Optional, Union
from sqlalchemy.orm import Session
from app.core.security import get_password_hash, verify_password
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
def get_user(db: Session, user_id: int) -> Optional[User]:
return db.query(User).filter(User.id == user_id).first()
def get_user_by_email(db: Session, email: str) -> Optional[User]:
return db.query(User).filter(User.email == email).first()
def get_users(
db: Session, skip: int = 0, limit: int = 100
) -> list[User]:
return db.query(User).offset(skip).limit(limit).all()
def create_user(db: Session, obj_in: UserCreate) -> User:
db_obj = User(
email=obj_in.email,
hashed_password=get_password_hash(obj_in.password),
is_active=True,
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def update_user(
db: Session, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]
) -> User:
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.model_dump(exclude_unset=True)
if update_data.get("password"):
hashed_password = get_password_hash(update_data["password"])
del update_data["password"]
update_data["hashed_password"] = hashed_password
for field in update_data:
setattr(db_obj, field, update_data[field])
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def authenticate(db: Session, *, email: str, password: str) -> Optional[User]:
user = get_user_by_email(db, email=email)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user

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()

5
app/db/base_class.py Normal file
View File

@ -0,0 +1,5 @@
# Import all the models, so that Base has them before being
# imported by Alembic
from app.db.base import Base # noqa
from app.models.user import User # noqa
from app.models.todo import Todo # noqa

53
app/db/deps.py Normal file
View File

@ -0,0 +1,53 @@
from typing import Generator
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt
from pydantic import ValidationError
from sqlalchemy.orm import Session
from app.core.config import settings
from app.db.session import SessionLocal
from app.models.user import User
from app.crud.crud_user import get_user
from app.schemas.token import TokenPayload
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl=f"{settings.API_V1_STR}/auth/login"
)
def get_db() -> Generator:
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_current_user(
db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> User:
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=["HS256"]
)
token_data = TokenPayload(**payload)
except (jwt.JWTError, ValidationError):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Could not validate credentials",
)
user = get_user(db, user_id=token_data.sub)
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
def get_current_active_user(
current_user: User = Depends(get_current_user),
) -> User:
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user

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

@ -0,0 +1,14 @@
from pathlib import Path
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
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)

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

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

@ -0,0 +1,16 @@
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, Text
from sqlalchemy.orm import relationship
from app.db.base import Base
class Todo(Base):
__tablename__ = "todos"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(Text, nullable=True)
is_completed = Column(Boolean, default=False)
owner_id = Column(Integer, ForeignKey("users.id"))
owner = relationship("User", back_populates="todos")

15
app/models/user.py Normal file
View File

@ -0,0 +1,15 @@
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import relationship
from app.db.base import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
is_active = Column(Boolean, default=True)
todos = relationship("Todo", back_populates="owner")

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

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

@ -0,0 +1,40 @@
from typing import Optional
from pydantic import BaseModel
# Shared properties
class TodoBase(BaseModel):
title: Optional[str] = None
description: Optional[str] = None
is_completed: Optional[bool] = False
# Properties to receive on item creation
class TodoCreate(TodoBase):
title: str
# Properties to receive on item update
class TodoUpdate(TodoBase):
pass
# Properties shared by models stored in DB
class TodoInDBBase(TodoBase):
id: int
title: str
owner_id: int
class Config:
from_attributes = True
# Properties to return to client
class Todo(TodoInDBBase):
pass
# Properties properties stored in DB
class TodoInDB(TodoInDBBase):
pass

12
app/schemas/token.py Normal file
View File

@ -0,0 +1,12 @@
from typing import Optional
from pydantic import BaseModel
class Token(BaseModel):
access_token: str
token_type: str
class TokenPayload(BaseModel):
sub: Optional[int] = None

37
app/schemas/user.py Normal file
View File

@ -0,0 +1,37 @@
from typing import Optional
from pydantic import BaseModel, EmailStr
# Shared properties
class UserBase(BaseModel):
email: Optional[EmailStr] = None
is_active: Optional[bool] = True
# Properties to receive via API on creation
class UserCreate(UserBase):
email: EmailStr
password: str
# Properties to receive via API on update
class UserUpdate(UserBase):
password: Optional[str] = None
class UserInDBBase(UserBase):
id: Optional[int] = None
class Config:
from_attributes = True
# Additional properties to return via API
class User(UserInDBBase):
pass
# Additional properties stored in DB
class UserInDB(UserInDBBase):
hashed_password: str

40
main.py Normal file
View File

@ -0,0 +1,40 @@
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1.api import api_router
from app.core.config import settings
app = FastAPI(
title=settings.PROJECT_NAME,
openapi_url="/openapi.json",
)
# Set all CORS enabled origins
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(api_router, prefix=settings.API_V1_STR)
@app.get("/")
async def root():
return {
"title": settings.PROJECT_NAME,
"docs_url": "/docs",
"health_check": "/health"
}
@app.get("/health", status_code=200)
async def health_check():
return {"status": "healthy"}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

0
migrations/__init__.py Normal file
View File

90
migrations/env.py Normal file
View File

@ -0,0 +1,90 @@
import os
import sys
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
# Interpret the config file for Python logging.
# This line sets up loggers basically.
fileConfig(config.config_file_name)
# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
# Make sure the app module is accessible
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, base_dir)
# Import Base from base_class which includes all models
from app.db.base_class import Base # noqa
target_metadata = Base.metadata
# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.
def run_migrations_offline():
"""Run migrations in 'offline' mode.
This configures the context with just a URL
and not an Engine, though an Engine is acceptable
here as well. By skipping the Engine creation
we don't even need a DBAPI to be available.
Calls to context.execute() here emit the given string to the
script output.
"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online():
"""Run migrations in 'online' mode.
In this scenario we need to create an Engine
and associate a connection with the context.
"""
connectable = engine_from_config(
config.get_section(config.config_ini_section),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
is_sqlite = connection.dialect.name == 'sqlite'
context.configure(
connection=connection,
target_metadata=target_metadata,
render_as_batch=is_sqlite,
compare_type=True,
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

24
migrations/script.py.mako Normal file
View File

@ -0,0 +1,24 @@
"""${message}
Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}
# revision identifiers, used by Alembic.
revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}
def upgrade():
${upgrades if upgrades else "pass"}
def downgrade():
${downgrades if downgrades else "pass"}

View File

@ -0,0 +1,53 @@
"""initial migration
Revision ID: 001
Revises:
Create Date: 2023-11-15
"""
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 users table
op.create_table(
'users',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('email', sa.String(), nullable=True),
sa.Column('hashed_password', sa.String(), nullable=True),
sa.Column('is_active', sa.Boolean(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False)
# Create todos table
op.create_table(
'todos',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('title', sa.String(), nullable=True),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('is_completed', sa.Boolean(), nullable=True),
sa.Column('owner_id', sa.Integer(), nullable=True),
sa.ForeignKeyConstraint(['owner_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_todos_id'), 'todos', ['id'], unique=False)
op.create_index(op.f('ix_todos_title'), 'todos', ['title'], unique=False)
def downgrade():
op.drop_index(op.f('ix_todos_title'), table_name='todos')
op.drop_index(op.f('ix_todos_id'), table_name='todos')
op.drop_table('todos')
op.drop_index(op.f('ix_users_id'), table_name='users')
op.drop_index(op.f('ix_users_email'), table_name='users')
op.drop_table('users')

View File

11
requirements.txt Normal file
View File

@ -0,0 +1,11 @@
fastapi==0.104.1
pydantic==2.4.2
pydantic-settings==2.0.3
sqlalchemy==2.0.23
alembic==1.12.1
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.6
uvicorn==0.24.0
python-dotenv==1.0.0
ruff==0.1.5