Implement calories calculator API with FastAPI

This commit is contained in:
Automated Action 2025-05-30 20:35:55 +00:00
parent 3162a7d094
commit 2f6fcf68fe
40 changed files with 2271 additions and 2 deletions

134
README.md
View File

@ -1,3 +1,133 @@
# FastAPI Application
# Calories Calculator API
This is a FastAPI application bootstrapped by BackendIM, the AI-powered backend generation platform.
A FastAPI-based API for tracking calories and nutritional information. This application allows users to register, track their food intake, and calculate their recommended calorie intake based on personal metrics.
## Features
- User registration and authentication
- Food database management
- Calorie tracking by meal type
- Daily and weekly calorie summaries
- BMR (Basal Metabolic Rate) and TDEE (Total Daily Energy Expenditure) calculations
- Personalized calorie recommendations based on goals
## Getting Started
### Prerequisites
- Python 3.8+
- SQLite
### Installation
1. Clone the repository:
```bash
git clone <repository-url>
cd caloriescalculatorapi
```
2. Install dependencies:
```bash
pip install -r requirements.txt
```
3. Run database migrations:
```bash
alembic upgrade head
```
4. Start the application:
```bash
uvicorn main:app --reload
```
The API will be available at http://localhost:8000.
### API Documentation
Once the application is running, you can access the interactive API documentation at:
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
## API Endpoints
### Authentication
- `POST /api/v1/auth/register` - Register a new user
- `POST /api/v1/auth/login` - Login and get access token
- `POST /api/v1/auth/test-token` - Test if the token is valid
### Users
- `GET /api/v1/users/me` - Get current user data
- `PUT /api/v1/users/me` - Update current user data
### Foods
- `GET /api/v1/foods` - List food items
- `POST /api/v1/foods` - Create a new food item
- `GET /api/v1/foods/{food_id}` - Get a specific food item
- `PUT /api/v1/foods/{food_id}` - Update a food item
- `DELETE /api/v1/foods/{food_id}` - Delete a food item
- `GET /api/v1/foods/my-foods` - List food items created by the current user
### Calorie Entries
- `GET /api/v1/calorie-entries` - List calorie entries
- `POST /api/v1/calorie-entries` - Create a new calorie entry
- `GET /api/v1/calorie-entries/{entry_id}` - Get a specific calorie entry
- `PUT /api/v1/calorie-entries/{entry_id}` - Update a calorie entry
- `DELETE /api/v1/calorie-entries/{entry_id}` - Delete a calorie entry
- `GET /api/v1/calorie-entries/daily-summary` - Get daily calorie summary
- `GET /api/v1/calorie-entries/weekly-summary` - Get weekly calorie summary
### Calculator
- `POST /api/v1/calculator/calculate` - Calculate recommended calories
- `GET /api/v1/calculator/calculate-from-profile` - Calculate recommended calories based on user profile
## Project Structure
```
app/
├── api/ # API endpoints
│ └── v1/
│ ├── endpoints/ # API route handlers
│ └── api.py # API router configuration
├── core/ # Core functionality
│ ├── config.py # Application configuration
│ ├── deps.py # Dependencies
│ └── security.py # Security utilities
├── db/ # Database configuration
│ ├── base.py # SQLAlchemy Base class
│ └── session.py # Database session setup
├── models/ # SQLAlchemy models
├── schemas/ # Pydantic schemas
├── services/ # Business logic
└── utils/ # Utility functions
migrations/ # Alembic migrations
main.py # Application entry point
```
## Development
### Running Tests
```bash
pytest
```
### Running Linters
```bash
ruff check .
```
## License
This project is licensed under the MIT License.

84
alembic.ini Normal file
View File

@ -0,0 +1,84 @@
# 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

3
app/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""
App package initialization
"""

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

@ -0,0 +1,3 @@
"""
API package initialization
"""

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

@ -0,0 +1,3 @@
"""
API v1 package initialization
"""

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

@ -0,0 +1,13 @@
"""
API router configuration
"""
from fastapi import APIRouter
from app.api.v1.endpoints import auth, users, foods, calorie_entries, calculator
api_router = APIRouter()
api_router.include_router(auth.router, prefix="/auth", tags=["authentication"])
api_router.include_router(users.router, prefix="/users", tags=["users"])
api_router.include_router(foods.router, prefix="/foods", tags=["foods"])
api_router.include_router(calorie_entries.router, prefix="/calorie-entries", tags=["calorie-entries"])
api_router.include_router(calculator.router, prefix="/calculator", tags=["calculator"])

View File

@ -0,0 +1,3 @@
"""
API v1 endpoints package initialization
"""

View File

@ -0,0 +1,76 @@
"""
Authentication endpoints
"""
from datetime import timedelta
from typing import Any
from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from app import models, schemas, services
from app.core import security
from app.core.config import settings
from app.core.deps import get_current_user
from app.db.session import get_db
router = APIRouter()
@router.post("/login", response_model=schemas.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 = services.user.authenticate(
db, email_or_username=form_data.username, password=form_data.password
)
if not user:
raise HTTPException(status_code=400, detail="Incorrect email/username or password")
elif not services.user.is_active(user):
raise HTTPException(status_code=400, detail="Inactive user")
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return {
"access_token": security.create_access_token(
user.id, expires_delta=access_token_expires
),
"token_type": "bearer",
}
@router.post("/test-token", response_model=schemas.User)
def test_token(current_user: models.User = Depends(get_current_user)) -> Any:
"""
Test access token
"""
return current_user
@router.post("/register", response_model=schemas.User)
def register_user(
*,
db: Session = Depends(get_db),
user_in: schemas.UserCreate,
) -> Any:
"""
Register a new user
"""
user = services.user.get_by_email(db, email=user_in.email)
if user:
raise HTTPException(
status_code=400,
detail="A user with this email already exists",
)
user = services.user.get_by_username(db, username=user_in.username)
if user:
raise HTTPException(
status_code=400,
detail="A user with this username already exists",
)
user = services.user.create(db, obj_in=user_in)
return user

View File

@ -0,0 +1,78 @@
"""
Calorie calculator endpoints
"""
from datetime import date
from typing import Any, Dict
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app import models
from app.core.deps import get_current_active_user
from app.db.session import get_db
from app.utils.calories import (
ActivityLevel,
WeightGoal,
calculate_recommended_calories
)
router = APIRouter()
class CalorieCalculationRequest(BaseModel):
"""Request model for calorie calculation"""
weight_kg: float
height_cm: float
birth_date: date
gender: str
activity_level: ActivityLevel
goal: WeightGoal
@router.post("/calculate", response_model=Dict[str, Any])
def calculate_calories(
*,
calculation_in: CalorieCalculationRequest,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Calculate recommended daily calories based on personal data and goals.
"""
result = calculate_recommended_calories(
weight_kg=calculation_in.weight_kg,
height_cm=calculation_in.height_cm,
birth_date=calculation_in.birth_date,
gender=calculation_in.gender,
activity_level=calculation_in.activity_level,
goal=calculation_in.goal
)
return result
@router.get("/calculate-from-profile", response_model=Dict[str, Any])
def calculate_calories_from_profile(
*,
db: Session = Depends(get_db),
activity_level: ActivityLevel,
goal: WeightGoal,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Calculate recommended daily calories based on user profile data.
"""
if not current_user.weight_kg or not current_user.height_cm or not current_user.date_of_birth or not current_user.gender:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Profile data incomplete. Please update your profile with weight, height, date of birth, and gender."
)
result = calculate_recommended_calories(
weight_kg=current_user.weight_kg,
height_cm=current_user.height_cm,
birth_date=current_user.date_of_birth,
gender=current_user.gender,
activity_level=activity_level,
goal=goal
)
return result

View File

@ -0,0 +1,231 @@
"""
Calorie entries endpoints
"""
from datetime import date
from typing import Any, Dict, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from app import models, schemas, services
from app.core.deps import get_current_active_user
from app.db.session import get_db
router = APIRouter()
@router.get("/", response_model=List[schemas.CalorieEntryWithFood])
def read_entries(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
start_date: Optional[date] = None,
end_date: Optional[date] = None,
meal_type: Optional[str] = None,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve calorie entries for the current user with optional filtering.
"""
entries = services.calorie_entry.get_multi_by_user(
db,
user_id=current_user.id,
skip=skip,
limit=limit,
start_date=start_date,
end_date=end_date,
meal_type=meal_type,
)
# Calculate calories for each entry
result = []
for entry in entries:
calories = entry.quantity_g / 100 * entry.food.calories_per_100g
# Create a CalorieEntryWithFood object
entry_with_food = schemas.CalorieEntryWithFood(
id=entry.id,
user_id=entry.user_id,
food_id=entry.food_id,
quantity_g=entry.quantity_g,
meal_type=entry.meal_type,
notes=entry.notes,
consumed_at=entry.consumed_at,
created_at=entry.created_at,
updated_at=entry.updated_at,
food=schemas.Food.model_validate(entry.food),
calories=round(calories, 1),
)
result.append(entry_with_food)
return result
@router.post("/", response_model=schemas.CalorieEntry)
def create_entry(
*,
db: Session = Depends(get_db),
entry_in: schemas.CalorieEntryCreate,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Create new calorie entry.
"""
# Verify food exists
food = services.food.get(db=db, food_id=entry_in.food_id)
if not food:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Food item not found",
)
entry = services.calorie_entry.create(
db=db, obj_in=entry_in, user_id=current_user.id
)
return entry
@router.get("/daily-summary", response_model=Dict[str, Any])
def get_daily_summary(
date_: date = Query(None, description="Date to get summary for, defaults to today"),
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Get calorie summary for a specific date. Defaults to today.
"""
if date_ is None:
date_ = date.today()
summary = services.calorie_entry.get_daily_summary(
db, user_id=current_user.id, date_=date_
)
return summary
@router.get("/weekly-summary", response_model=List[Dict[str, Any]])
def get_weekly_summary(
end_date: date = Query(None, description="End date for weekly summary, defaults to today"),
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Get calorie summary for the last 7 days ending on the specified date. Defaults to today.
"""
if end_date is None:
end_date = date.today()
summary = services.calorie_entry.get_weekly_summary(
db, user_id=current_user.id, end_date=end_date
)
return summary
@router.get("/{entry_id}", response_model=schemas.CalorieEntryWithFood)
def read_entry(
*,
db: Session = Depends(get_db),
entry_id: int,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Get calorie entry by ID.
"""
entry = services.calorie_entry.get(db=db, entry_id=entry_id)
if not entry:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Calorie entry not found",
)
# Verify ownership
if entry.user_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
# Calculate calories
calories = entry.quantity_g / 100 * entry.food.calories_per_100g
# Create a CalorieEntryWithFood object
entry_with_food = schemas.CalorieEntryWithFood(
id=entry.id,
user_id=entry.user_id,
food_id=entry.food_id,
quantity_g=entry.quantity_g,
meal_type=entry.meal_type,
notes=entry.notes,
consumed_at=entry.consumed_at,
created_at=entry.created_at,
updated_at=entry.updated_at,
food=schemas.Food.model_validate(entry.food),
calories=round(calories, 1),
)
return entry_with_food
@router.put("/{entry_id}", response_model=schemas.CalorieEntry)
def update_entry(
*,
db: Session = Depends(get_db),
entry_id: int,
entry_in: schemas.CalorieEntryUpdate,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Update a calorie entry.
"""
entry = services.calorie_entry.get(db=db, entry_id=entry_id)
if not entry:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Calorie entry not found",
)
# Verify ownership
if entry.user_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
# Verify food exists if changing food_id
if entry_in.food_id and entry_in.food_id != entry.food_id:
food = services.food.get(db=db, food_id=entry_in.food_id)
if not food:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Food item not found",
)
entry = services.calorie_entry.update(db=db, db_obj=entry, obj_in=entry_in)
return entry
@router.delete("/{entry_id}", response_model=schemas.CalorieEntry)
def delete_entry(
*,
db: Session = Depends(get_db),
entry_id: int,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Delete a calorie entry.
"""
entry = services.calorie_entry.get(db=db, entry_id=entry_id)
if not entry:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Calorie entry not found",
)
# Verify ownership
if entry.user_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
entry = services.calorie_entry.delete(db=db, entry_id=entry_id)
return entry

View File

@ -0,0 +1,165 @@
"""
Food items endpoints
"""
from typing import Any, List, Optional
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app import models, schemas, services
from app.core.deps import get_current_active_superuser, get_current_active_user
from app.db.session import get_db
router = APIRouter()
@router.get("/", response_model=List[schemas.Food])
def read_foods(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
name: Optional[str] = None,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve food items. Supports filtering by name.
"""
foods = services.food.get_multi(db, skip=skip, limit=limit, name_filter=name)
return foods
@router.post("/", response_model=schemas.Food)
def create_food(
*,
db: Session = Depends(get_db),
food_in: schemas.FoodCreate,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Create new food item.
"""
# Set is_verified to False for user-created foods, unless user is superuser
if not current_user.is_superuser:
food_in.is_verified = False
food = services.food.create(
db=db, obj_in=food_in, created_by_id=current_user.id
)
return food
@router.get("/my-foods", response_model=List[schemas.Food])
def read_user_foods(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Retrieve food items created by the current user.
"""
foods = services.food.get_multi_by_user(
db=db, user_id=current_user.id, skip=skip, limit=limit
)
return foods
@router.get("/{food_id}", response_model=schemas.Food)
def read_food(
*,
db: Session = Depends(get_db),
food_id: int,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Get food item by ID.
"""
food = services.food.get(db=db, food_id=food_id)
if not food:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Food item not found",
)
return food
@router.put("/{food_id}", response_model=schemas.Food)
def update_food(
*,
db: Session = Depends(get_db),
food_id: int,
food_in: schemas.FoodUpdate,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Update a food item.
"""
food = services.food.get(db=db, food_id=food_id)
if not food:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Food item not found",
)
# Only allow update if user is superuser or the creator of the food item
if not current_user.is_superuser and food.created_by_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
# Regular users cannot mark items as verified
if not current_user.is_superuser and food_in.is_verified:
food_in.is_verified = False
food = services.food.update(db=db, db_obj=food, obj_in=food_in)
return food
@router.delete("/{food_id}", response_model=schemas.Food)
def delete_food(
*,
db: Session = Depends(get_db),
food_id: int,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Delete a food item.
"""
food = services.food.get(db=db, food_id=food_id)
if not food:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Food item not found",
)
# Only allow deletion if user is superuser or the creator of the food item
if not current_user.is_superuser and food.created_by_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
food = services.food.delete(db=db, food_id=food_id)
return food
@router.post("/verify/{food_id}", response_model=schemas.Food)
def verify_food(
*,
db: Session = Depends(get_db),
food_id: int,
current_user: models.User = Depends(get_current_active_superuser),
) -> Any:
"""
Verify a food item. Only superusers can verify food items.
"""
food = services.food.get(db=db, food_id=food_id)
if not food:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Food item not found",
)
food = services.food.update(db=db, db_obj=food, obj_in={"is_verified": True})
return food

View File

@ -0,0 +1,131 @@
"""
User management endpoints
"""
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app import models, schemas, services
from app.core.deps import get_current_active_superuser, get_current_active_user
from app.db.session import get_db
router = APIRouter()
@router.get("/", response_model=List[schemas.User])
def read_users(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
current_user: models.User = Depends(get_current_active_superuser),
) -> Any:
"""
Retrieve users. Only superusers can access this endpoint.
"""
users = services.user.get_multi(db, skip=skip, limit=limit)
return users
@router.get("/me", response_model=schemas.User)
def read_user_me(
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Get current user.
"""
return current_user
@router.put("/me", response_model=schemas.User)
def update_user_me(
*,
db: Session = Depends(get_db),
user_in: schemas.UserUpdate,
current_user: models.User = Depends(get_current_active_user),
) -> Any:
"""
Update own user.
"""
if user_in.username and user_in.username != current_user.username:
user = services.user.get_by_username(db, username=user_in.username)
if user:
raise HTTPException(
status_code=400,
detail="Username already registered",
)
if user_in.email and user_in.email != current_user.email:
user = services.user.get_by_email(db, email=user_in.email)
if user:
raise HTTPException(
status_code=400,
detail="Email already registered",
)
user = services.user.update(db, db_obj=current_user, obj_in=user_in)
return user
@router.get("/{user_id}", response_model=schemas.User)
def read_user(
user_id: int,
current_user: models.User = Depends(get_current_active_user),
db: Session = Depends(get_db),
) -> Any:
"""
Get a specific user by id.
"""
user = services.user.get(db, user_id=user_id)
if user == current_user:
return user
if not services.user.is_superuser(current_user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="The user doesn't have enough privileges",
)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
return user
@router.put("/{user_id}", response_model=schemas.User)
def update_user(
*,
db: Session = Depends(get_db),
user_id: int,
user_in: schemas.UserUpdate,
current_user: models.User = Depends(get_current_active_superuser),
) -> Any:
"""
Update a user. Only superusers can access this endpoint.
"""
user = services.user.get(db, user_id=user_id)
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
# Check if username or email already exists
if user_in.username and user_in.username != user.username:
existing_user = services.user.get_by_username(db, username=user_in.username)
if existing_user:
raise HTTPException(
status_code=400,
detail="Username already registered",
)
if user_in.email and user_in.email != user.email:
existing_user = services.user.get_by_email(db, email=user_in.email)
if existing_user:
raise HTTPException(
status_code=400,
detail="Email already registered",
)
user = services.user.update(db, db_obj=user, obj_in=user_in)
return user

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

@ -0,0 +1,3 @@
"""
Core package initialization
"""

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

@ -0,0 +1,40 @@
"""
Application configuration settings
"""
from typing import List, Union
from pydantic import AnyHttpUrl, EmailStr, validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""Application settings"""
API_V1_STR: str = "/api/v1"
SECRET_KEY: str = "development_secret_key_change_in_production"
# 60 minutes * 24 hours * 8 days = 8 days
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8
# CORS Configuration
BACKEND_CORS_ORIGINS: List[AnyHttpUrl] = []
@validator("BACKEND_CORS_ORIGINS", pre=True)
def assemble_cors_origins(cls, v: Union[str, List[str]]) -> Union[List[str], str]:
"""Validate CORS origins"""
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)
# SQLite Database settings
SQLALCHEMY_DATABASE_URI: str = "sqlite:///app/storage/db/db.sqlite"
# Superuser settings for initial setup
FIRST_SUPERUSER: EmailStr = "admin@example.com"
FIRST_SUPERUSER_USERNAME: str = "admin"
FIRST_SUPERUSER_PASSWORD: str = "admin"
model_config = SettingsConfigDict(case_sensitive=True)
settings = Settings()

64
app/core/deps.py Normal file
View File

@ -0,0 +1,64 @@
"""
Dependency utilities for the application
"""
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from pydantic import ValidationError
from sqlalchemy.orm import Session
from app import models, schemas
from app.core.config import settings
from app.db.session import get_db
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl=f"{settings.API_V1_STR}/auth/login"
)
def get_current_user(
db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> models.User:
"""
Get the current user from the JWT token
"""
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=["HS256"]
)
token_data = schemas.TokenPayload(**payload)
except (JWTError, ValidationError):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Could not validate credentials",
)
user = db.query(models.User).filter(models.User.id == token_data.sub).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
def get_current_active_user(
current_user: models.User = Depends(get_current_user),
) -> models.User:
"""
Get the current active user
"""
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
def get_current_active_superuser(
current_user: models.User = Depends(get_current_user),
) -> models.User:
"""
Get the current active superuser
"""
if not current_user.is_superuser:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="The user doesn't have enough privileges",
)
return current_user

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

@ -0,0 +1,43 @@
"""
Security utilities for authentication
"""
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:
"""
Create a JWT access token
"""
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:
"""
Verify a password against a hash
"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""
Hash a password
"""
return pwd_context.hash(password)

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

@ -0,0 +1,31 @@
"""
Database Base models configuration
"""
from typing import Any
from sqlalchemy import MetaData
from sqlalchemy.ext.declarative import as_declarative, declared_attr
convention = {
"ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s",
}
metadata = MetaData(naming_convention=convention)
@as_declarative(metadata=metadata)
class Base:
"""Base class for all database models"""
id: Any
__name__: str
# Generate __tablename__ automatically based on class name
@declared_attr
def __tablename__(cls) -> str:
return cls.__name__.lower()

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

@ -0,0 +1,30 @@
"""
Database session configuration
"""
from pathlib import Path
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# Create the database directory 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(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False},
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
"""
Dependency for getting the database session
"""
db = SessionLocal()
try:
yield db
finally:
db.close()

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

@ -0,0 +1,8 @@
"""
Models package initialization
"""
from app.models.user import User
from app.models.food import Food
from app.models.calorie_entry import CalorieEntry
__all__ = ["User", "Food", "CalorieEntry"]

View File

@ -0,0 +1,25 @@
"""
CalorieEntry model for the application
"""
from datetime import datetime
from sqlalchemy import Column, String, Integer, DateTime, Float, ForeignKey
from sqlalchemy.orm import relationship
from app.db.base import Base
class CalorieEntry(Base):
"""CalorieEntry model for tracking user's calorie intake"""
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("user.id"), nullable=False)
food_id = Column(Integer, ForeignKey("food.id"), nullable=False)
quantity_g = Column(Float, nullable=False)
meal_type = Column(String, nullable=True) # breakfast, lunch, dinner, snack
notes = Column(String, nullable=True)
consumed_at = Column(DateTime, default=datetime.utcnow)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
user = relationship("User", back_populates="calorie_entries")
food = relationship("Food", back_populates="calorie_entries")

29
app/models/food.py Normal file
View File

@ -0,0 +1,29 @@
"""
Food model for the application
"""
from datetime import datetime
from sqlalchemy import Column, String, Integer, DateTime, Float, Boolean, ForeignKey
from sqlalchemy.orm import relationship
from app.db.base import Base
class Food(Base):
"""Food model for storing food items and their nutritional information"""
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True, nullable=False)
description = Column(String, nullable=True)
calories_per_100g = Column(Float, nullable=False)
protein_g = Column(Float, nullable=True)
carbs_g = Column(Float, nullable=True)
fat_g = Column(Float, nullable=True)
fiber_g = Column(Float, nullable=True)
is_verified = Column(Boolean, default=False)
# If created by user, store the user ID
created_by_id = Column(Integer, ForeignKey("user.id"), nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
created_by = relationship("User", backref="created_foods")
calorie_entries = relationship("CalorieEntry", back_populates="food")

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

@ -0,0 +1,31 @@
"""
User model for the application
"""
from datetime import datetime
from sqlalchemy import Column, String, Integer, DateTime, Float, Boolean
from sqlalchemy.orm import relationship
from app.db.base import Base
class User(Base):
"""User model for storing user related data"""
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)
first_name = Column(String, nullable=True)
last_name = Column(String, nullable=True)
is_active = Column(Boolean, default=True)
is_superuser = Column(Boolean, default=False)
# User's target daily calories
target_calories = Column(Float, default=2000.0)
height_cm = Column(Float, nullable=True)
weight_kg = Column(Float, nullable=True)
date_of_birth = Column(DateTime, nullable=True)
gender = Column(String, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationships
calorie_entries = relationship("CalorieEntry", back_populates="user", cascade="all, delete-orphan")

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

@ -0,0 +1,14 @@
"""
Schemas package initialization
"""
from app.schemas.user import User, UserCreate, UserUpdate, UserInDB
from app.schemas.food import Food, FoodCreate, FoodUpdate
from app.schemas.calorie_entry import CalorieEntry, CalorieEntryCreate, CalorieEntryUpdate, CalorieEntryWithFood
from app.schemas.token import Token, TokenPayload
__all__ = [
"User", "UserCreate", "UserUpdate", "UserInDB",
"Food", "FoodCreate", "FoodUpdate",
"CalorieEntry", "CalorieEntryCreate", "CalorieEntryUpdate", "CalorieEntryWithFood",
"Token", "TokenPayload"
]

17
app/schemas/base.py Normal file
View File

@ -0,0 +1,17 @@
"""
Base schemas for the application
"""
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, ConfigDict
class BaseSchema(BaseModel):
"""Base schema with common fields and config"""
model_config = ConfigDict(from_attributes=True)
class TimestampSchema(BaseSchema):
"""Schema with timestamp fields"""
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None

View File

@ -0,0 +1,52 @@
"""
CalorieEntry schemas for the application
"""
from datetime import datetime
from typing import Optional
from pydantic import Field
from app.schemas.base import BaseSchema, TimestampSchema
from app.schemas.food import Food
# Shared properties
class CalorieEntryBase(BaseSchema):
"""Base calorie entry schema with shared properties"""
food_id: Optional[int] = None
quantity_g: Optional[float] = None
meal_type: Optional[str] = None
notes: Optional[str] = None
consumed_at: Optional[datetime] = None
# Properties to receive via API on creation
class CalorieEntryCreate(CalorieEntryBase):
"""Schema for creating a new calorie entry"""
food_id: int
quantity_g: float = Field(..., gt=0)
consumed_at: Optional[datetime] = datetime.utcnow()
# Properties to receive via API on update
class CalorieEntryUpdate(CalorieEntryBase):
"""Schema for updating a calorie entry"""
pass
class CalorieEntryInDBBase(CalorieEntryBase, TimestampSchema):
"""Base schema for calorie entry in database"""
id: int
user_id: int
# Additional properties to return via API
class CalorieEntry(CalorieEntryInDBBase):
"""Schema for returning a calorie entry via API"""
pass
# Expanded calorie entry with food details
class CalorieEntryWithFood(CalorieEntry):
"""Schema for returning a calorie entry with food details"""
food: Food
calories: float # Calculated total calories for this entry

45
app/schemas/food.py Normal file
View File

@ -0,0 +1,45 @@
"""
Food schemas for the application
"""
from typing import Optional
from pydantic import Field
from app.schemas.base import BaseSchema, TimestampSchema
# Shared properties
class FoodBase(BaseSchema):
"""Base food schema with shared properties"""
name: Optional[str] = None
description: Optional[str] = None
calories_per_100g: Optional[float] = None
protein_g: Optional[float] = None
carbs_g: Optional[float] = None
fat_g: Optional[float] = None
fiber_g: Optional[float] = None
is_verified: Optional[bool] = False
# Properties to receive via API on creation
class FoodCreate(FoodBase):
"""Schema for creating a new food item"""
name: str
calories_per_100g: float = Field(..., gt=0)
# Properties to receive via API on update
class FoodUpdate(FoodBase):
"""Schema for updating a food item"""
pass
class FoodInDBBase(FoodBase, TimestampSchema):
"""Base schema for food in database"""
id: int
created_by_id: Optional[int] = None
# Additional properties to return via API
class Food(FoodInDBBase):
"""Schema for returning a food item via API"""
pass

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

@ -0,0 +1,17 @@
"""
Token schemas for authentication
"""
from typing import Optional
from pydantic import BaseModel
class Token(BaseModel):
"""Schema for access token"""
access_token: str
token_type: str = "bearer"
class TokenPayload(BaseModel):
"""Schema for token payload"""
sub: Optional[int] = None # Subject (user ID)
exp: Optional[int] = None # Expiration time

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

@ -0,0 +1,55 @@
"""
User schemas for the application
"""
from datetime import datetime
from typing import Optional
from pydantic import EmailStr
from app.schemas.base import BaseSchema, TimestampSchema
# Shared properties
class UserBase(BaseSchema):
"""Base user schema with shared properties"""
email: Optional[EmailStr] = None
username: Optional[str] = None
is_active: Optional[bool] = True
is_superuser: bool = False
first_name: Optional[str] = None
last_name: Optional[str] = None
target_calories: Optional[float] = 2000.0
height_cm: Optional[float] = None
weight_kg: Optional[float] = None
date_of_birth: Optional[datetime] = None
gender: Optional[str] = None
# Properties to receive via API on creation
class UserCreate(UserBase):
"""Schema for creating a new user"""
email: EmailStr
username: str
password: str
# Properties to receive via API on update
class UserUpdate(UserBase):
"""Schema for updating a user"""
password: Optional[str] = None
class UserInDBBase(UserBase, TimestampSchema):
"""Base schema for user in database"""
id: int
# Additional properties to return via API
class User(UserInDBBase):
"""Schema for returning a user via API"""
pass
# Additional properties stored in DB but not returned by API
class UserInDB(UserInDBBase):
"""Schema for user in database with hashed password"""
hashed_password: str

3
app/services/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""
Services package initialization
"""

View File

@ -0,0 +1,210 @@
"""
CalorieEntry service for managing calorie tracking operations
"""
from datetime import datetime, date, timedelta
from typing import Optional, List, Dict, Any, Union
from sqlalchemy import func
from sqlalchemy.orm import Session, joinedload
from app import models, schemas
def get(db: Session, entry_id: int) -> Optional[models.CalorieEntry]:
"""
Get a calorie entry by ID
"""
return (
db.query(models.CalorieEntry)
.filter(models.CalorieEntry.id == entry_id)
.options(joinedload(models.CalorieEntry.food))
.first()
)
def get_multi_by_user(
db: Session,
*,
user_id: int,
skip: int = 0,
limit: int = 100,
start_date: Optional[date] = None,
end_date: Optional[date] = None,
meal_type: Optional[str] = None,
) -> List[models.CalorieEntry]:
"""
Get multiple calorie entries for a specific user with optional filters
"""
query = (
db.query(models.CalorieEntry)
.filter(models.CalorieEntry.user_id == user_id)
.options(joinedload(models.CalorieEntry.food))
)
if start_date:
query = query.filter(
func.date(models.CalorieEntry.consumed_at) >= start_date
)
if end_date:
query = query.filter(
func.date(models.CalorieEntry.consumed_at) <= end_date
)
if meal_type:
query = query.filter(models.CalorieEntry.meal_type == meal_type)
return (
query.order_by(models.CalorieEntry.consumed_at.desc())
.offset(skip)
.limit(limit)
.all()
)
def get_daily_summary(
db: Session,
*,
user_id: int,
date_: date
) -> Dict[str, Any]:
"""
Get a summary of calories for a specific date
"""
start_datetime = datetime.combine(date_, datetime.min.time())
end_datetime = datetime.combine(date_, datetime.max.time())
entries = (
db.query(models.CalorieEntry)
.join(models.Food, models.CalorieEntry.food_id == models.Food.id)
.filter(models.CalorieEntry.user_id == user_id)
.filter(models.CalorieEntry.consumed_at >= start_datetime)
.filter(models.CalorieEntry.consumed_at <= end_datetime)
.options(joinedload(models.CalorieEntry.food))
.all()
)
# Calculate total calories and nutrients
total_calories = 0.0
total_protein = 0.0
total_carbs = 0.0
total_fat = 0.0
total_fiber = 0.0
meal_summary = {}
for entry in entries:
# Calculate calories for this entry
calories = entry.quantity_g / 100 * entry.food.calories_per_100g
total_calories += calories
# Calculate nutrients if available
if entry.food.protein_g:
total_protein += entry.quantity_g / 100 * entry.food.protein_g
if entry.food.carbs_g:
total_carbs += entry.quantity_g / 100 * entry.food.carbs_g
if entry.food.fat_g:
total_fat += entry.quantity_g / 100 * entry.food.fat_g
if entry.food.fiber_g:
total_fiber += entry.quantity_g / 100 * entry.food.fiber_g
# Add to meal summary
meal_type = entry.meal_type or "Other"
if meal_type not in meal_summary:
meal_summary[meal_type] = 0.0
meal_summary[meal_type] += calories
# Get user's target calories
user = db.query(models.User).filter(models.User.id == user_id).first()
target_calories = user.target_calories if user else 2000.0
return {
"date": date_,
"total_calories": round(total_calories, 1),
"target_calories": target_calories,
"remaining_calories": round(target_calories - total_calories, 1),
"total_protein_g": round(total_protein, 1),
"total_carbs_g": round(total_carbs, 1),
"total_fat_g": round(total_fat, 1),
"total_fiber_g": round(total_fiber, 1),
"meal_summary": {k: round(v, 1) for k, v in meal_summary.items()},
"entry_count": len(entries),
}
def create(
db: Session, *, obj_in: schemas.CalorieEntryCreate, user_id: int
) -> models.CalorieEntry:
"""
Create a new calorie entry
"""
db_obj = models.CalorieEntry(
user_id=user_id,
food_id=obj_in.food_id,
quantity_g=obj_in.quantity_g,
meal_type=obj_in.meal_type,
notes=obj_in.notes,
consumed_at=obj_in.consumed_at or datetime.utcnow(),
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def update(
db: Session,
*,
db_obj: models.CalorieEntry,
obj_in: Union[schemas.CalorieEntryUpdate, Dict[str, Any]]
) -> models.CalorieEntry:
"""
Update a calorie entry
"""
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.model_dump(exclude_unset=True)
for field in update_data:
if hasattr(db_obj, field):
setattr(db_obj, field, update_data[field])
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def delete(db: Session, *, entry_id: int) -> models.CalorieEntry:
"""
Delete a calorie entry
"""
entry = db.query(models.CalorieEntry).filter(models.CalorieEntry.id == entry_id).first()
db.delete(entry)
db.commit()
return entry
def get_weekly_summary(
db: Session,
*,
user_id: int,
end_date: date = None
) -> List[Dict[str, Any]]:
"""
Get a weekly summary of calories
"""
if end_date is None:
end_date = date.today()
start_date = end_date - timedelta(days=6) # Last 7 days
summaries = []
current_date = start_date
while current_date <= end_date:
daily_summary = get_daily_summary(db, user_id=user_id, date_=current_date)
summaries.append(daily_summary)
current_date += timedelta(days=1)
return summaries

99
app/services/food.py Normal file
View File

@ -0,0 +1,99 @@
"""
Food service for managing food operations
"""
from typing import Optional, List, Dict, Any, Union
from sqlalchemy.orm import Session
from app import models, schemas
def get(db: Session, food_id: int) -> Optional[models.Food]:
"""
Get a food item by ID
"""
return db.query(models.Food).filter(models.Food.id == food_id).first()
def get_multi(
db: Session, *, skip: int = 0, limit: int = 100, name_filter: Optional[str] = None
) -> List[models.Food]:
"""
Get multiple food items, with optional name filter
"""
query = db.query(models.Food)
if name_filter:
query = query.filter(models.Food.name.ilike(f"%{name_filter}%"))
return query.order_by(models.Food.name).offset(skip).limit(limit).all()
def get_multi_by_user(
db: Session, *, user_id: int, skip: int = 0, limit: int = 100
) -> List[models.Food]:
"""
Get multiple food items created by a specific user
"""
return (
db.query(models.Food)
.filter(models.Food.created_by_id == user_id)
.order_by(models.Food.name)
.offset(skip)
.limit(limit)
.all()
)
def create(
db: Session, *, obj_in: schemas.FoodCreate, created_by_id: Optional[int] = None
) -> models.Food:
"""
Create a new food item
"""
db_obj = models.Food(
name=obj_in.name,
description=obj_in.description,
calories_per_100g=obj_in.calories_per_100g,
protein_g=obj_in.protein_g,
carbs_g=obj_in.carbs_g,
fat_g=obj_in.fat_g,
fiber_g=obj_in.fiber_g,
is_verified=obj_in.is_verified,
created_by_id=created_by_id,
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def update(
db: Session, *, db_obj: models.Food, obj_in: Union[schemas.FoodUpdate, Dict[str, Any]]
) -> models.Food:
"""
Update a food item
"""
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.model_dump(exclude_unset=True)
for field in update_data:
if hasattr(db_obj, field):
setattr(db_obj, field, update_data[field])
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def delete(db: Session, *, food_id: int) -> models.Food:
"""
Delete a food item
"""
food = db.query(models.Food).filter(models.Food.id == food_id).first()
db.delete(food)
db.commit()
return food

126
app/services/user.py Normal file
View File

@ -0,0 +1,126 @@
"""
User service for managing user operations
"""
from typing import Optional, List, Dict, Any, Union
from sqlalchemy.orm import Session
from app import models, schemas
from app.core.security import get_password_hash, verify_password
def get_by_email(db: Session, email: str) -> Optional[models.User]:
"""
Get a user by email
"""
return db.query(models.User).filter(models.User.email == email).first()
def get_by_username(db: Session, username: str) -> Optional[models.User]:
"""
Get a user by username
"""
return db.query(models.User).filter(models.User.username == username).first()
def get(db: Session, user_id: int) -> Optional[models.User]:
"""
Get a user by ID
"""
return db.query(models.User).filter(models.User.id == user_id).first()
def get_multi(
db: Session, *, skip: int = 0, limit: int = 100
) -> List[models.User]:
"""
Get multiple users
"""
return db.query(models.User).offset(skip).limit(limit).all()
def create(db: Session, *, obj_in: schemas.UserCreate) -> models.User:
"""
Create a new user
"""
db_obj = models.User(
email=obj_in.email,
username=obj_in.username,
hashed_password=get_password_hash(obj_in.password),
first_name=obj_in.first_name,
last_name=obj_in.last_name,
target_calories=obj_in.target_calories,
height_cm=obj_in.height_cm,
weight_kg=obj_in.weight_kg,
date_of_birth=obj_in.date_of_birth,
gender=obj_in.gender,
is_superuser=obj_in.is_superuser,
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def update(
db: Session, *, db_obj: models.User, obj_in: Union[schemas.UserUpdate, Dict[str, Any]]
) -> models.User:
"""
Update a 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:
if hasattr(db_obj, field):
setattr(db_obj, field, update_data[field])
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def delete(db: Session, *, user_id: int) -> models.User:
"""
Delete a user
"""
user = db.query(models.User).filter(models.User.id == user_id).first()
db.delete(user)
db.commit()
return user
def authenticate(db: Session, *, email_or_username: str, password: str) -> Optional[models.User]:
"""
Authenticate a user with email/username and password
"""
user = get_by_email(db, email=email_or_username)
if not user:
user = get_by_username(db, username=email_or_username)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user
def is_active(user: models.User) -> bool:
"""
Check if a user is active
"""
return user.is_active
def is_superuser(user: models.User) -> bool:
"""
Check if a user is a superuser
"""
return user.is_superuser

3
app/utils/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""
Utils package initialization
"""

132
app/utils/calories.py Normal file
View File

@ -0,0 +1,132 @@
"""
Utility functions for calories calculation
"""
from datetime import date
from enum import Enum
from typing import Dict, Any
class ActivityLevel(str, Enum):
"""Activity level for BMR calculations"""
SEDENTARY = "sedentary" # Little or no exercise
LIGHTLY_ACTIVE = "lightly_active" # Light exercise/sports 1-3 days/week
MODERATELY_ACTIVE = "moderately_active" # Moderate exercise/sports 3-5 days/week
VERY_ACTIVE = "very_active" # Hard exercise/sports 6-7 days/week
EXTRA_ACTIVE = "extra_active" # Very hard exercise, physical job or training twice a day
class WeightGoal(str, Enum):
"""Weight goals for calorie adjustments"""
LOSE_FAST = "lose_fast" # Lose 1kg per week
LOSE = "lose" # Lose 0.5kg per week
MAINTAIN = "maintain" # Maintain current weight
GAIN = "gain" # Gain 0.5kg per week
GAIN_FAST = "gain_fast" # Gain 1kg per week
def calculate_age(birth_date: date) -> int:
"""
Calculate age from birth date
"""
today = date.today()
age = today.year - birth_date.year
# Check if birthday has occurred this year
if today.month < birth_date.month or (today.month == birth_date.month and today.day < birth_date.day):
age -= 1
return age
def calculate_bmr(
weight_kg: float,
height_cm: float,
age: int,
gender: str
) -> float:
"""
Calculate Basal Metabolic Rate (BMR) using the Mifflin-St Jeor Equation
For men: BMR = (10 × weight in kg) + (6.25 × height in cm) - (5 × age in years) + 5
For women: BMR = (10 × weight in kg) + (6.25 × height in cm) - (5 × age in years) - 161
"""
if gender.lower() in ("male", "m"):
return (10 * weight_kg) + (6.25 * height_cm) - (5 * age) + 5
else: # female
return (10 * weight_kg) + (6.25 * height_cm) - (5 * age) - 161
def calculate_tdee(bmr: float, activity_level: ActivityLevel) -> float:
"""
Calculate Total Daily Energy Expenditure (TDEE) based on BMR and activity level
"""
activity_multipliers = {
ActivityLevel.SEDENTARY: 1.2,
ActivityLevel.LIGHTLY_ACTIVE: 1.375,
ActivityLevel.MODERATELY_ACTIVE: 1.55,
ActivityLevel.VERY_ACTIVE: 1.725,
ActivityLevel.EXTRA_ACTIVE: 1.9,
}
return bmr * activity_multipliers[activity_level]
def adjust_calories_for_goal(tdee: float, goal: WeightGoal) -> float:
"""
Adjust TDEE based on weight goal
1kg of fat is approximately 7700 calories
To lose/gain 1kg per week: 7700 / 7 = 1100 calories per day
To lose/gain 0.5kg per week: 7700 / 14 = 550 calories per day
"""
goal_adjustments = {
WeightGoal.LOSE_FAST: -1100,
WeightGoal.LOSE: -550,
WeightGoal.MAINTAIN: 0,
WeightGoal.GAIN: 550,
WeightGoal.GAIN_FAST: 1100,
}
return tdee + goal_adjustments[goal]
def calculate_recommended_calories(
weight_kg: float,
height_cm: float,
birth_date: date,
gender: str,
activity_level: ActivityLevel,
goal: WeightGoal
) -> Dict[str, Any]:
"""
Calculate recommended daily calories based on personal data and goals
"""
age = calculate_age(birth_date)
bmr = calculate_bmr(weight_kg, height_cm, age, gender)
tdee = calculate_tdee(bmr, activity_level)
adjusted_calories = adjust_calories_for_goal(tdee, goal)
# Calculate macronutrient recommendations based on a balanced diet
# Protein: 30%, Carbs: 40%, Fat: 30%
calories_from_protein = adjusted_calories * 0.3
calories_from_carbs = adjusted_calories * 0.4
calories_from_fat = adjusted_calories * 0.3
# Convert calories to grams
# Protein: 4 calories per gram
# Carbs: 4 calories per gram
# Fat: 9 calories per gram
protein_g = calories_from_protein / 4
carbs_g = calories_from_carbs / 4
fat_g = calories_from_fat / 9
return {
"bmr": round(bmr),
"tdee": round(tdee),
"recommended_calories": round(adjusted_calories),
"macronutrients": {
"protein_g": round(protein_g),
"carbs_g": round(carbs_g),
"fat_g": round(fat_g),
}
}

55
main.py Normal file
View File

@ -0,0 +1,55 @@
"""
Calories Calculator API
"""
import pathlib
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from app.api.v1.api import api_router
from app.core.config import settings
# Create database directory if it doesn't exist
DB_DIR = pathlib.Path("/app") / "storage" / "db"
DB_DIR.mkdir(parents=True, exist_ok=True)
app = FastAPI(
title="Calories Calculator API",
description="API for tracking calories and nutritional information",
version="0.1.0",
openapi_url="/openapi.json",
)
# CORS middleware setup
app.add_middleware(
CORSMiddleware,
allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS],
allow_credentials=True,
allow_methods=["*"], # Allows all methods
allow_headers=["*"], # Allows all headers
)
# Include API router
app.include_router(api_router, prefix=settings.API_V1_STR)
@app.get("/health", tags=["Health"])
async def health_check():
"""
Health check endpoint for the API.
"""
return JSONResponse({"status": "healthy"})
@app.get("/openapi.json", include_in_schema=False)
async def get_open_api_endpoint():
"""
Return the OpenAPI schema for the API.
"""
return app.openapi()
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

3
migrations/__init__.py Normal file
View File

@ -0,0 +1,3 @@
"""
Migrations package initialization
"""

85
migrations/env.py Normal file
View File

@ -0,0 +1,85 @@
"""
Alembic environment configuration
"""
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
# Import models to ensure they are registered with the Base metadata
import app.models # noqa: F401
# 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
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, # Set batch mode for 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,93 @@
"""Initial tables
Revision ID: 001
Revises:
Create Date: 2023-09-26 10:00:00.000000
"""
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('first_name', sa.String(), nullable=True),
sa.Column('last_name', sa.String(), nullable=True),
sa.Column('is_active', sa.Boolean(), nullable=True, default=True),
sa.Column('is_superuser', sa.Boolean(), nullable=True, default=False),
sa.Column('target_calories', sa.Float(), nullable=True, default=2000.0),
sa.Column('height_cm', sa.Float(), nullable=True),
sa.Column('weight_kg', sa.Float(), nullable=True),
sa.Column('date_of_birth', sa.DateTime(), nullable=True),
sa.Column('gender', sa.String(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=True),
sa.Column('updated_at', sa.DateTime(), 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 food table
op.create_table(
'food',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('description', sa.String(), nullable=True),
sa.Column('calories_per_100g', sa.Float(), nullable=False),
sa.Column('protein_g', sa.Float(), nullable=True),
sa.Column('carbs_g', sa.Float(), nullable=True),
sa.Column('fat_g', sa.Float(), nullable=True),
sa.Column('fiber_g', sa.Float(), nullable=True),
sa.Column('is_verified', sa.Boolean(), nullable=True, default=False),
sa.Column('created_by_id', sa.Integer(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=True),
sa.Column('updated_at', sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(['created_by_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_food_id'), 'food', ['id'], unique=False)
op.create_index(op.f('ix_food_name'), 'food', ['name'], unique=False)
# Create calorie_entry table
op.create_table(
'calorieentry',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('food_id', sa.Integer(), nullable=False),
sa.Column('quantity_g', sa.Float(), nullable=False),
sa.Column('meal_type', sa.String(), nullable=True),
sa.Column('notes', sa.String(), nullable=True),
sa.Column('consumed_at', sa.DateTime(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=True),
sa.Column('updated_at', sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(['food_id'], ['food.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_calorieentry_id'), 'calorieentry', ['id'], unique=False)
def downgrade():
op.drop_index(op.f('ix_calorieentry_id'), table_name='calorieentry')
op.drop_table('calorieentry')
op.drop_index(op.f('ix_food_name'), table_name='food')
op.drop_index(op.f('ix_food_id'), table_name='food')
op.drop_table('food')
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')

12
requirements.txt Normal file
View File

@ -0,0 +1,12 @@
fastapi>=0.103.1
uvicorn>=0.23.2
pydantic>=2.3.0
pydantic-settings>=2.0.3
pydantic[email]>=2.3.0
sqlalchemy>=2.0.20
alembic>=1.12.0
python-jose[cryptography]>=3.3.0
passlib[bcrypt]>=1.7.4
python-multipart>=0.0.6
email-validator>=2.0.0
ruff>=0.0.292