backend.im/taskmanagerapi-4423u0
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement Task Manager API with FastAPI and SQLite

Browse Source
This commit is contained in:
Automated Action 2025-06-11 17:21:30 +00:00
parent 80960ab8a3
commit 35910a1b62
33 changed files with 1258 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

134
README.md
View File

@ -1,3 +1,133 @@
# FastAPI Application
# Task Manager API
This is a FastAPI application bootstrapped by BackendIM, the AI-powered backend generation platform.
A FastAPI-based Task Manager API with SQLite database and JWT authentication.
## Features
- User management (register, login, update profile)
- Task management (create, read, update, delete)
- Task assignment to other users
- Task filtering by status
- JWT authentication
- Role-based access control (regular users and superusers)
## Tech Stack
- **Framework**: FastAPI
- **Database**: SQLite with SQLAlchemy ORM
- **Migrations**: Alembic
- **Authentication**: JWT tokens (via python-jose)
- **Password Hashing**: Bcrypt
- **Validation**: Pydantic
## Project Structure
```
├── app/
│ ├── api/
│ │ └── v1/
│ │ ├── endpoints/
│ │ │ ├── auth.py
│ │ │ ├── tasks.py
│ │ │ └── users.py
│ │ └── api.py
│ ├── core/
│ │ ├── auth.py
│ │ └── config.py
│ ├── crud/
│ │ ├── task.py
│ │ └── user.py
│ ├── db/
│ │ ├── base_class.py
│ │ ├── base.py
│ │ └── session.py
│ ├── models/
│ │ ├── task.py
│ │ └── user.py
│ ├── schemas/
│ │ ├── task.py
│ │ ├── token.py
│ │ └── user.py
│ └── storage/
│ └── db/
├── migrations/
│ └── versions/
├── alembic.ini
├── main.py
├── README.md
└── requirements.txt
```
## Environment Variables
This application uses the following environment variables:
| Variable | Description | Default |
|----------|-------------|---------|
| SECRET_KEY | JWT secret key | supersecretkey |
| ACCESS_TOKEN_EXPIRE_MINUTES | JWT token expiration time in minutes | 60 |
## Getting Started
### Prerequisites
- Python 3.8 or higher
### Installation
1. Clone the repository
2. Install dependencies
```bash
pip install -r requirements.txt
```
3. Run the application
```bash
uvicorn main:app --reload
```
4. Access the API documentation
- Swagger UI: [http://localhost:8000/docs](http://localhost:8000/docs)
- ReDoc: [http://localhost:8000/redoc](http://localhost:8000/redoc)
### Database Migrations
Apply migrations to create the database schema:
```bash
alembic upgrade head
```
## API Endpoints
### Authentication
- `POST /api/v1/auth/register` - Register a new user
- `POST /api/v1/auth/token` - Login and get access token
### Users
- `GET /api/v1/users/me` - Get current user
- `PUT /api/v1/users/me` - Update current user
- `GET /api/v1/users/{user_id}` - Get user by ID
- `GET /api/v1/users/` - List all users (superuser only)
- `POST /api/v1/users/` - Create a new user (superuser only)
- `PUT /api/v1/users/{user_id}` - Update a user (superuser only)
### Tasks
- `GET /api/v1/tasks/` - List tasks (with optional status filter)
- `POST /api/v1/tasks/` - Create a new task
- `GET /api/v1/tasks/{task_id}` - Get a task by ID
- `PUT /api/v1/tasks/{task_id}` - Update a task
- `DELETE /api/v1/tasks/{task_id}` - Delete a task
- `POST /api/v1/tasks/{task_id}/assign/{user_id}` - Assign a task to a user
### Health Check
- `GET /health` - Check API health
## License
This project is licensed under the MIT License.

111
alembic.ini Normal file
View File

@ -0,0 +1,111 @@
# A generic, single database configuration.
[alembic]
# path to migration scripts
script_location = migrations
# template used to generate migration file names; The default value is %%(rev)s_%%(slug)s
# Uncomment the line below if you want the files to be prepended with date and time
# see https://alembic.sqlalchemy.org/en/latest/tutorial.html#editing-the-ini-file
# for all available tokens
# file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s
# sys.path path, will be prepended to sys.path if present.
# defaults to the current working directory.
prepend_sys_path = .
# timezone to use when rendering the date within the migration file
# as well as the filename.
# If specified, requires the python-dateutil library that can be
# installed by adding `alembic[tz]` to the pip requirements
# 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.
# The path separator used here should be the separator specified by "version_path_separator" below.
# version_locations = %(here)s/bar:%(here)s/bat:migrations/versions
# version path separator; As mentioned above, this is the character used to split
# version_locations. The default within new alembic.ini files is "os", which uses os.pathsep.
# If this key is omitted entirely, it falls back to the legacy behavior of splitting on spaces and/or commas.
# Valid values for version_path_separator are:
#
# version_path_separator = :
# version_path_separator = ;
# version_path_separator = space
version_path_separator = os # Use os.pathsep. Default configuration used for new projects.
# set to 'true' to search source files recursively
# in each "version_locations" directory
# new in Alembic version 1.10
# recursive_version_locations = false
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
# SQLite URL - using absolute path as recommended
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 REVISION_SCRIPT_FILENAME
# 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

1
app/__init__.py Normal file
View File

@ -0,0 +1 @@
# Empty __init__.py file to make the directory a Python package

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

@ -0,0 +1 @@
# Empty __init__.py file to make the directory a Python package

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

@ -0,0 +1 @@
# Empty __init__.py file to make the directory a Python package

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

@ -0,0 +1,9 @@
from fastapi import APIRouter
from app.api.v1.endpoints import tasks, users, auth
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(tasks.router, prefix="/tasks", tags=["tasks"])

1
app/api/v1/endpoints/__init__.py Normal file
View File

@ -0,0 +1 @@
# Empty __init__.py file to make the directory a Python package

51
app/api/v1/endpoints/auth.py Normal file
View File

@ -0,0 +1,51 @@
from datetime import timedelta
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.core.auth import authenticate_user, create_access_token
from app.core.config import settings
from app.schemas.token import Token
from app.schemas.user import UserCreate, UserResponse
from app.crud import user as crud_user
router = APIRouter()
@router.post("/token", response_model=Token)
async def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
"""
OAuth2 compatible token login, get an access token for future requests.
"""
user = authenticate_user(db, email=form_data.username, password=form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.email}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def register_user(
user_in: UserCreate,
db: Session = Depends(get_db)
):
"""
Register a new user.
"""
user = crud_user.get_user_by_email(db, email=user_in.email)
if user:
raise HTTPException(
status_code=400,
detail="The user with this email already exists in the system",
)
user = crud_user.create_user(db=db, obj_in=user_in)
return user

120
app/api/v1/endpoints/tasks.py Normal file
View File

@ -0,0 +1,120 @@
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.models.user import User
from app.core.auth import get_current_user
from app.schemas.task import TaskCreate, TaskUpdate, TaskResponse
from app.crud import task as crud_task
router = APIRouter()
@router.post("/", response_model=TaskResponse, status_code=status.HTTP_201_CREATED)
def create_task(
task_in: TaskCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Create a new task for the current user.
"""
task = crud_task.create_task(db=db, task_in=task_in, owner_id=current_user.id)
return task
@router.get("/", response_model=List[TaskResponse])
def read_tasks(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
status: Optional[str] = None
):
"""
Retrieve all tasks for the current user.
Optional status filter available.
"""
tasks = crud_task.get_tasks_by_owner(
db=db,
owner_id=current_user.id,
skip=skip,
limit=limit,
status=status
)
return tasks
@router.get("/{task_id}", response_model=TaskResponse)
def read_task(
task_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Get a specific task by ID.
"""
task = crud_task.get_task(db=db, task_id=task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
if task.owner_id != current_user.id and task.assigned_to_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
return task
@router.put("/{task_id}", response_model=TaskResponse)
def update_task(
task_id: int,
task_in: TaskUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Update a task.
"""
task = crud_task.get_task(db=db, task_id=task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
if task.owner_id != current_user.id and task.assigned_to_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
task = crud_task.update_task(db=db, db_obj=task, obj_in=task_in)
return task
@router.delete("/{task_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
def delete_task(
task_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Delete a task.
"""
task = crud_task.get_task(db=db, task_id=task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
if task.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
crud_task.delete_task(db=db, task_id=task_id)
return None
@router.post("/{task_id}/assign/{user_id}", response_model=TaskResponse)
def assign_task(
task_id: int,
user_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Assign a task to another user.
"""
task = crud_task.get_task(db=db, task_id=task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
if task.owner_id != current_user.id:
raise HTTPException(status_code=403, detail="Not enough permissions")
# Check if user exists
from app.crud.user import get_user
user = get_user(db=db, user_id=user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
task = crud_task.assign_task(db=db, task_id=task_id, assigned_to_id=user_id)
return task

105
app/api/v1/endpoints/users.py Normal file
View File

@ -0,0 +1,105 @@
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.user import User
from app.core.auth import get_current_user, get_current_active_superuser
from app.schemas.user import UserCreate, UserUpdate, UserResponse
from app.crud import user as crud_user
router = APIRouter()
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
def create_user(
user_in: UserCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_superuser)
):
"""
Create new user (superuser only).
"""
user = crud_user.get_user_by_email(db, email=user_in.email)
if user:
raise HTTPException(
status_code=400,
detail="The user with this email already exists in the system."
)
user = crud_user.create_user(db=db, obj_in=user_in)
return user
@router.get("/", response_model=List[UserResponse])
def read_users(
skip: int = 0,
limit: int = 100,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_superuser)
):
"""
Retrieve users (superuser only).
"""
users = crud_user.get_users(db, skip=skip, limit=limit)
return users
@router.get("/me", response_model=UserResponse)
def read_user_me(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Get current user.
"""
return current_user
@router.put("/me", response_model=UserResponse)
def update_user_me(
user_in: UserUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Update own user.
"""
user = crud_user.update_user(db, db_obj=current_user, obj_in=user_in)
return user
@router.get("/{user_id}", response_model=UserResponse)
def read_user_by_id(
user_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user)
):
"""
Get a specific user by id.
"""
user = crud_user.get_user(db, user_id=user_id)
if not user:
raise HTTPException(
status_code=404,
detail="The user with this ID does not exist in the system"
)
if user.id != current_user.id and not current_user.is_superuser:
raise HTTPException(
status_code=403,
detail="Not enough permissions"
)
return user
@router.put("/{user_id}", response_model=UserResponse)
def update_user(
user_id: int,
user_in: UserUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_active_superuser)
):
"""
Update a user (superuser only).
"""
user = crud_user.get_user(db, user_id=user_id)
if not user:
raise HTTPException(
status_code=404,
detail="The user with this ID does not exist in the system"
)
user = crud_user.update_user(db, db_obj=user, obj_in=user_in)
return user

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

@ -0,0 +1 @@
# Empty __init__.py file to make the directory a Python package

107
app/core/auth.py Normal file
View File

@ -0,0 +1,107 @@
from datetime import datetime, timedelta
from typing import Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from passlib.context import CryptContext
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.core.config import settings
from app.models.user import User
from app.schemas.token import TokenPayload
# Password hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# OAuth2 token URL
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/token")
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""
Verify that a plain password matches a hashed password.
"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""
Hash a password for storing in the database.
"""
return pwd_context.hash(password)
def authenticate_user(db: Session, email: str, password: str) -> Optional[User]:
"""
Try to authenticate the user.
"""
from app.crud.user import get_user_by_email
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
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
"""
Create JWT access token.
"""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
) -> User:
"""
Get the current user from the token.
"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
)
email: str = payload.get("sub")
if email is None:
raise credentials_exception
token_data = TokenPayload(email=email)
except JWTError:
raise credentials_exception
from app.crud.user import get_user_by_email
user = get_user_by_email(db, email=token_data.email)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(
current_user: User = Depends(get_current_user),
) -> User:
"""
Get the current active user.
"""
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
async def get_current_active_superuser(
current_user: User = Depends(get_current_user),
) -> User:
"""
Get the current active superuser.
"""
if not current_user.is_superuser:
raise HTTPException(
status_code=403, detail="The user doesn't have enough privileges"
)
return current_user

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

@ -0,0 +1,25 @@
import os
from pathlib import Path
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
PROJECT_NAME: str = "Task Manager API"
PROJECT_DESCRIPTION: str = "A FastAPI-based Task Manager API"
VERSION: str = "0.1.0"
# Database
DB_DIR: Path = Path("/app") / "storage" / "db"
DATABASE_URL: str = f"sqlite:///{DB_DIR}/db.sqlite"
# Security
SECRET_KEY: str = os.getenv("SECRET_KEY", "supersecretkey")
ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60"))
ALGORITHM: str = "HS256"
model_config = SettingsConfigDict(env_file=".env", case_sensitive=True)
# Create an instance of the Settings class
settings = Settings()
# Ensure DB directory exists
settings.DB_DIR.mkdir(parents=True, exist_ok=True)

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

@ -0,0 +1 @@
# Import crud modules here to make them importable from 'app.crud'

90
app/crud/task.py Normal file
View File

@ -0,0 +1,90 @@
from typing import List, Optional, Union, Dict, Any
from sqlalchemy.orm import Session
from app.models.task import Task
from app.schemas.task import TaskCreate, TaskUpdate
def get_task(db: Session, task_id: int) -> Optional[Task]:
"""
Get a task by ID.
"""
return db.query(Task).filter(Task.id == task_id).first()
def get_tasks_by_owner(
db: Session,
owner_id: int,
skip: int = 0,
limit: int = 100,
status: Optional[str] = None
) -> List[Task]:
"""
Get tasks by owner ID with optional status filter.
"""
query = db.query(Task).filter(Task.owner_id == owner_id)
if status:
query = query.filter(Task.status == status)
return query.offset(skip).limit(limit).all()
def create_task(db: Session, task_in: TaskCreate, owner_id: int) -> Task:
"""
Create a new task.
"""
task = Task(
**task_in.dict(),
owner_id=owner_id
)
db.add(task)
db.commit()
db.refresh(task)
return task
def update_task(
db: Session,
db_obj: Task,
obj_in: Union[TaskUpdate, Dict[str, Any]]
) -> Task:
"""
Update a task.
"""
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.dict(exclude_unset=True)
for field, value in update_data.items():
if value is not None and hasattr(db_obj, field):
setattr(db_obj, field, value)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def delete_task(db: Session, task_id: int) -> None:
"""
Delete a task.
"""
task = db.query(Task).filter(Task.id == task_id).first()
if task:
db.delete(task)
db.commit()
def assign_task(db: Session, task_id: int, assigned_to_id: int) -> Task:
"""
Assign a task to a user.
"""
task = get_task(db, task_id=task_id)
task.assigned_to_id = assigned_to_id
db.add(task)
db.commit()
db.refresh(task)
return task

71
app/crud/user.py Normal file
View File

@ -0,0 +1,71 @@
from typing import List, Optional, Union, Dict, Any
from sqlalchemy.orm import Session
from app.core.auth import get_password_hash
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
def get_user(db: Session, user_id: int) -> Optional[User]:
"""
Get a user by ID.
"""
return db.query(User).filter(User.id == user_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.
"""
db_obj = User(
email=obj_in.email,
hashed_password=get_password_hash(obj_in.password),
full_name=obj_in.full_name,
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: Union[UserUpdate, Dict[str, Any]]
) -> User:
"""
Update a user.
"""
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.dict(exclude_unset=True)
if "password" in update_data and update_data["password"]:
hashed_password = get_password_hash(update_data["password"])
del update_data["password"]
update_data["hashed_password"] = hashed_password
for field, value in update_data.items():
if value is not None and hasattr(db_obj, field):
setattr(db_obj, field, value)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj

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

@ -0,0 +1 @@
# Empty __init__.py file to make the directory a Python package

5
app/db/base.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_class import Base # noqa
from app.models.user import User # noqa
from app.models.task import Task # noqa

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

@ -0,0 +1,13 @@
from typing import Any
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import DeclarativeBase
class Base(DeclarativeBase):
id: Any
__name__: str
# Generate __tablename__ automatically
@declared_attr
def __tablename__(cls) -> str:
return cls.__name__.lower()

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

@ -0,0 +1,25 @@
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
# Create SQLAlchemy engine
engine = create_engine(
settings.DATABASE_URL,
connect_args={"check_same_thread": False} # Needed for SQLite
)
# Create a SessionLocal class
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Dependency to get DB session
def get_db():
"""
Dependency function that yields db sessions
"""
db = SessionLocal()
try:
yield db
finally:
db.close()

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

@ -0,0 +1 @@
# Empty __init__.py file to make the directory a Python package

36
app/models/task.py Normal file
View File

@ -0,0 +1,36 @@
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, Enum
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
import enum
from app.db.base_class import Base
class TaskStatus(str, enum.Enum):
TODO = "todo"
IN_PROGRESS = "in_progress"
DONE = "done"
class TaskPriority(str, enum.Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
class Task(Base):
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True, nullable=False)
description = Column(Text, nullable=True)
status = Column(Enum(TaskStatus), default=TaskStatus.TODO, nullable=False)
priority = Column(Enum(TaskPriority), default=TaskPriority.MEDIUM, nullable=False)
due_date = Column(DateTime(timezone=True), nullable=True)
# Foreign keys
owner_id = Column(Integer, ForeignKey("user.id"), nullable=False)
assigned_to_id = Column(Integer, ForeignKey("user.id"), nullable=True)
# Timestamps
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Relationships
owner = relationship("User", foreign_keys=[owner_id], back_populates="tasks_owned")
assigned_to = relationship("User", foreign_keys=[assigned_to_id], back_populates="tasks_assigned")

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

@ -0,0 +1,20 @@
from sqlalchemy import Boolean, Column, Integer, String, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.sql import func
from app.db.base_class import Base
class User(Base):
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
full_name = Column(String, nullable=True)
hashed_password = Column(String, nullable=False)
is_active = Column(Boolean, default=True)
is_superuser = Column(Boolean, default=False)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())
# Relationships
tasks_owned = relationship("Task", foreign_keys="Task.owner_id", back_populates="owner", cascade="all, delete-orphan")
tasks_assigned = relationship("Task", foreign_keys="Task.assigned_to_id", back_populates="assigned_to")

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

@ -0,0 +1 @@
# Empty __init__.py file to make the directory a Python package

36
app/schemas/task.py Normal file
View File

@ -0,0 +1,36 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, ConfigDict
from app.models.task import TaskStatus, TaskPriority
class TaskBase(BaseModel):
title: str
description: Optional[str] = None
status: TaskStatus = TaskStatus.TODO
priority: TaskPriority = TaskPriority.MEDIUM
due_date: Optional[datetime] = None
class TaskCreate(TaskBase):
assigned_to_id: Optional[int] = None
class TaskUpdate(BaseModel):
title: Optional[str] = None
description: Optional[str] = None
status: Optional[TaskStatus] = None
priority: Optional[TaskPriority] = None
due_date: Optional[datetime] = None
assigned_to_id: Optional[int] = None
class TaskResponse(TaskBase):
id: int
owner_id: int
assigned_to_id: Optional[int] = None
created_at: datetime
updated_at: Optional[datetime] = None
model_config = ConfigDict(from_attributes=True)

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

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

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

@ -0,0 +1,30 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, EmailStr, ConfigDict
class UserBase(BaseModel):
email: EmailStr
full_name: Optional[str] = None
is_active: bool = True
is_superuser: bool = False
class UserCreate(UserBase):
password: str
class UserUpdate(BaseModel):
email: Optional[EmailStr] = None
full_name: Optional[str] = None
password: Optional[str] = None
is_active: Optional[bool] = None
is_superuser: Optional[bool] = None
class UserResponse(UserBase):
id: int
created_at: datetime
updated_at: Optional[datetime] = None
model_config = ConfigDict(from_attributes=True)

49
main.py Normal file
View File

@ -0,0 +1,49 @@
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,
description=settings.PROJECT_DESCRIPTION,
version=settings.VERSION,
openapi_url="/openapi.json",
docs_url="/docs",
redoc_url="/redoc",
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include routers
app.include_router(api_router, prefix="/api/v1")
@app.get("/")
async def root():
"""
Root endpoint returning basic information about the service.
"""
return {
"title": settings.PROJECT_NAME,
"description": settings.PROJECT_DESCRIPTION,
"docs": "/docs",
"health": "/health",
}
@app.get("/health", tags=["health"])
async def health_check():
"""
Health check endpoint.
"""
return {"status": "healthy", "version": settings.VERSION}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

1
migrations/README Normal file
View File

@ -0,0 +1 @@
Generic single-database configuration with SQLite.

83
migrations/env.py Normal file
View File

@ -0,0 +1,83 @@
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
from app.db.base import Base
# 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.
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# add your model's MetaData object here
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
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() -> None:
"""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"},
render_as_batch=True, # SQLite-specific configuration
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""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, # SQLite-specific configuration
)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

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

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

72
migrations/versions/1_initial_migration.py Normal file
View File

@ -0,0 +1,72 @@
"""initial migration
Revision ID: 1_initial_migration
Revises:
Create Date: 2023-10-24 10:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '1_initial_migration'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# Create user table
op.create_table(
'user',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('email', sa.String(), nullable=False),
sa.Column('full_name', sa.String(), nullable=True),
sa.Column('hashed_password', sa.String(), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=True, default=True),
sa.Column('is_superuser', sa.Boolean(), nullable=True, default=False),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), 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)
# Create task table with TaskStatus and TaskPriority enums
op.create_table(
'task',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('title', sa.String(), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('status', sa.Enum('todo', 'in_progress', 'done', name='taskstatus'), nullable=False, default='todo'),
sa.Column('priority', sa.Enum('low', 'medium', 'high', name='taskpriority'), nullable=False, default='medium'),
sa.Column('due_date', sa.DateTime(timezone=True), nullable=True),
sa.Column('owner_id', sa.Integer(), nullable=False),
sa.Column('assigned_to_id', sa.Integer(), nullable=True),
sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=True),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=True),
sa.ForeignKeyConstraint(['owner_id'], ['user.id'], ),
sa.ForeignKeyConstraint(['assigned_to_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_task_id'), 'task', ['id'], unique=False)
op.create_index(op.f('ix_task_title'), 'task', ['title'], unique=False)
def downgrade() -> None:
# Drop task table
op.drop_index(op.f('ix_task_title'), table_name='task')
op.drop_index(op.f('ix_task_id'), table_name='task')
op.drop_table('task')
# Drop enums
op.execute('DROP TYPE IF EXISTS taskstatus')
op.execute('DROP TYPE IF EXISTS taskpriority')
# Drop user table
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')

20
requirements.txt Normal file
View File

@ -0,0 +1,20 @@
# FastAPI and web server
fastapi>=0.104.0
uvicorn>=0.23.2
# Database
SQLAlchemy>=2.0.0
alembic>=1.12.0
# Authentication
python-jose[cryptography]>=3.3.0
passlib[bcrypt]>=1.7.4
python-multipart>=0.0.6
# Validation and serialization
pydantic>=2.4.2
pydantic-settings>=2.0.3
email-validator>=2.0.0
# Development and linting
ruff>=0.1.0