Implement small business inventory management system

This commit is contained in:
Automated Action 2025-06-01 10:05:16 +00:00
parent fcf5da3665
commit d01f405afa
44 changed files with 2402 additions and 2 deletions

227
README.md
View File

@ -1,3 +1,226 @@
# FastAPI Application
# Small Business Inventory Management System
This is a FastAPI application bootstrapped by BackendIM, the AI-powered backend generation platform.
A FastAPI-based REST API for managing inventory in small businesses. This system helps small businesses track inventory, manage stock levels, categorize products, and generate reports.
## Features
- **User Authentication and Authorization**
- JWT-based authentication
- Role-based permissions (admin/regular users)
- Secure password hashing
- **Inventory Management**
- Add, update, and delete inventory items
- Track stock levels and set reorder points
- Assign items to categories
- Track item history with SKU
- **Category Management**
- Organize items into categories
- Nested categorization support
- Category-based reporting
- **Transaction Tracking**
- Record purchases, sales, returns, and adjustments
- Automatic stock level updates
- Transaction history for audit purposes
- **Reporting**
- Inventory value reports
- Category summary reports
- Low stock alerts
- Transaction summary reports
- **API Features**
- RESTful endpoints following best practices
- Comprehensive documentation with Swagger UI
- Health check endpoint for monitoring
## Tech Stack
- **FastAPI**: Modern, fast web framework for building APIs
- **SQLAlchemy**: SQL toolkit and ORM
- **Alembic**: Database migration tool
- **SQLite**: Simple, file-based database
- **Pydantic**: Data validation and settings management
- **Uvicorn**: ASGI server for FastAPI
- **Python-Jose**: JWT token handling
- **Passlib**: Password hashing
- **Ruff**: Python linter for code quality
## Project Structure
```
.
├── app/
│ ├── api/
│ │ ├── deps.py # Dependency injection
│ │ └── v1/
│ │ ├── api.py # API router
│ │ └── endpoints/
│ │ ├── auth.py # Authentication endpoints
│ │ ├── categories.py # Category endpoints
│ │ ├── health.py # Health check endpoint
│ │ ├── items.py # Inventory item endpoints
│ │ ├── reports.py # Reporting endpoints
│ │ ├── transactions.py # Transaction endpoints
│ │ └── users.py # User endpoints
│ ├── core/
│ │ ├── config.py # Application configuration
│ │ └── security.py # Security utilities
│ ├── crud/
│ │ ├── base.py # Base CRUD operations
│ │ ├── category.py # Category CRUD
│ │ ├── item.py # Item CRUD
│ │ ├── transaction.py # Transaction CRUD
│ │ └── user.py # User CRUD
│ ├── db/
│ │ ├── init_db.py # Database initialization
│ │ ├── session.py # Database session
│ │ └── utils.py # Database utilities
│ ├── models/ # SQLAlchemy models
│ │ ├── base.py # Base model
│ │ ├── category.py # Category model
│ │ ├── item.py # Item model
│ │ ├── transaction.py # Transaction model
│ │ └── user.py # User model
│ ├── schemas/ # Pydantic schemas
│ │ ├── category.py # Category schemas
│ │ ├── item.py # Item schemas
│ │ ├── report.py # Report schemas
│ │ ├── token.py # Token schemas
│ │ ├── transaction.py # Transaction schemas
│ │ └── user.py # User schemas
│ └── storage/ # Storage directory
│ └── db/ # Database storage
├── migrations/ # Alembic migrations
│ ├── env.py # Alembic environment
│ ├── script.py.mako # Migration script template
│ └── versions/ # Migration versions
│ └── initial_migration.py # Initial database schema
├── alembic.ini # Alembic configuration
├── main.py # Application entry point
├── pyproject.toml # Project configuration
└── requirements.txt # Dependencies
```
## Getting Started
### Prerequisites
- Python 3.8 or higher
### Installation
1. Clone the repository
2. Install dependencies:
```bash
pip install -r requirements.txt
```
3. Initialize the database:
```bash
alembic upgrade head
```
4. (Optional) Create a `.env` file in the root directory to customize environment variables:
```
PROJECT_NAME="My Inventory System"
SECRET_KEY="your-secret-key-here"
ACCESS_TOKEN_EXPIRE_MINUTES=1440 # 24 hours
```
5. Run the application:
```bash
uvicorn main:app --reload
```
6. Open http://localhost:8000/docs to access the Swagger UI documentation
## API Documentation
- **OpenAPI Documentation**: `/docs`
- Interactive API documentation with Swagger UI
- Test endpoints directly from the browser
- **Alternative Documentation**: `/redoc`
- Alternative documentation interface
- **OpenAPI JSON**: `/openapi.json`
- Raw OpenAPI specification
## API Endpoints
### Authentication
- `POST /api/v1/auth/login` - Obtain JWT token
- `POST /api/v1/auth/test-token` - Test token validity
### Users
- `GET /api/v1/users/` - List users (admin only)
- `POST /api/v1/users/` - Create user (admin only)
- `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
- `PUT /api/v1/users/{user_id}` - Update user (admin only)
### Categories
- `GET /api/v1/categories/` - List categories
- `POST /api/v1/categories/` - Create category (admin only)
- `GET /api/v1/categories/{id}` - Get category by ID
- `PUT /api/v1/categories/{id}` - Update category (admin only)
- `DELETE /api/v1/categories/{id}` - Delete category (admin only)
### Inventory Items
- `GET /api/v1/items/` - List items (with filtering options)
- `POST /api/v1/items/` - Create item
- `GET /api/v1/items/{id}` - Get item by ID
- `PUT /api/v1/items/{id}` - Update item
- `DELETE /api/v1/items/{id}` - Delete item
- `GET /api/v1/items/by-sku/{sku}` - Get item by SKU
- `GET /api/v1/items/low-stock/` - Get low stock items
### Transactions
- `GET /api/v1/transactions/` - List transactions (with filtering options)
- `POST /api/v1/transactions/` - Create transaction
- `GET /api/v1/transactions/{id}` - Get transaction by ID
### Reports
- `GET /api/v1/reports/inventory-value` - Inventory value report
- `GET /api/v1/reports/category-summary` - Category summary report
- `GET /api/v1/reports/low-stock` - Low stock report
- `GET /api/v1/reports/transaction-summary` - Transaction summary report
### Health Check
- `GET /health` - System health check
- `GET /api/v1/health` - Detailed health check with DB status
## Development
### Linting
The project uses Ruff for linting:
```bash
ruff check .
```
### Running Tests
```bash
pytest
```
## License
This project is licensed under the MIT License.

74
alembic.ini Normal file
View File

@ -0,0 +1,74 @@
# 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
# 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
# Use absolute path for SQLite URL
sqlalchemy.url = sqlite:////app/storage/db/db.sqlite

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

@ -0,0 +1,82 @@
from typing import Generator, Optional
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt
from pydantic import ValidationError
from sqlalchemy.orm import Session
from app.core.config import settings
from app.core.security import verify_password
from app.db.utils import get_db
from app.models.user import User
from app.schemas.token import TokenPayload
# OAuth2 configuration
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)
) -> User:
"""
Validate and decode the JWT token to get the current user.
"""
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
)
token_data = TokenPayload(**payload)
except (jwt.JWTError, ValidationError) as e:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Could not validate credentials",
) from e
user = db.query(User).filter(User.id == token_data.sub).first()
if not user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Inactive user",
)
return user
def get_current_active_superuser(
current_user: User = Depends(get_current_user),
) -> User:
"""
Check if the current user is a 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
def authenticate_user(
db: Session, email: str, password: str
) -> Optional[User]:
"""
Authenticate a user using email and password.
"""
user = db.query(User).filter(User.email == email).first()
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user

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

@ -0,0 +1,22 @@
from fastapi import APIRouter
from app.api.v1.endpoints import (
auth,
categories,
health,
items,
reports,
transactions,
users,
)
api_router = APIRouter()
# Include routers for different endpoints
api_router.include_router(auth.router, prefix="/auth", tags=["authentication"])
api_router.include_router(users.router, prefix="/users", tags=["users"])
api_router.include_router(items.router, prefix="/items", tags=["items"])
api_router.include_router(categories.router, prefix="/categories", tags=["categories"])
api_router.include_router(transactions.router, prefix="/transactions", tags=["transactions"])
api_router.include_router(reports.router, prefix="/reports", tags=["reports"])
api_router.include_router(health.router, prefix="/health", tags=["health"])

View File

@ -0,0 +1 @@
# Import endpoints to make them accessible

View File

@ -0,0 +1,52 @@
from datetime import timedelta
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from app.api.deps import authenticate_user
from app.core.config import settings
from app.core.security import create_access_token
from app.db.utils import get_db
from app.schemas.token import Token
router = APIRouter()
@router.post("/login", response_model=Token)
def login_access_token(
db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()
) -> Any:
"""
OAuth2 compatible token login, get an access token for future requests.
"""
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Inactive user",
)
access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
return {
"access_token": create_access_token(
user.id, expires_delta=access_token_expires
),
"token_type": "bearer",
}
@router.post("/test-token", response_model=None)
def test_token() -> Any:
"""
Test access token.
"""
return {"msg": "Token is valid"}

View File

@ -0,0 +1,117 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.api.deps import get_current_active_superuser, get_current_user
from app.crud import category
from app.db.utils import get_db
from app.models.user import User
from app.schemas.category import Category, CategoryCreate, CategoryUpdate
router = APIRouter()
@router.get("/", response_model=List[Category])
def read_categories(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Retrieve categories.
"""
categories = category.get_multi(db, skip=skip, limit=limit)
return categories
@router.post("/", response_model=Category)
def create_category(
*,
db: Session = Depends(get_db),
category_in: CategoryCreate,
current_user: User = Depends(get_current_active_superuser),
) -> Any:
"""
Create new category.
"""
db_category = category.get_by_name(db, name=category_in.name)
if db_category:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="The category with this name already exists in the system.",
)
return category.create(db, obj_in=category_in)
@router.get("/{id}", response_model=Category)
def read_category(
*,
db: Session = Depends(get_db),
id: int,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get category by ID.
"""
db_category = category.get(db, id=id)
if not db_category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found",
)
return db_category
@router.put("/{id}", response_model=Category)
def update_category(
*,
db: Session = Depends(get_db),
id: int,
category_in: CategoryUpdate,
current_user: User = Depends(get_current_active_superuser),
) -> Any:
"""
Update a category.
"""
db_category = category.get(db, id=id)
if not db_category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found",
)
# If name is being updated, check that it doesn't conflict with an existing category
if category_in.name and category_in.name != db_category.name:
existing_category = category.get_by_name(db, name=category_in.name)
if existing_category:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="The category with this name already exists in the system.",
)
return category.update(db, db_obj=db_category, obj_in=category_in)
@router.delete("/{id}", response_model=Category)
def delete_category(
*,
db: Session = Depends(get_db),
id: int,
current_user: User = Depends(get_current_active_superuser),
) -> Any:
"""
Delete a category.
"""
db_category = category.get(db, id=id)
if not db_category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found",
)
# Optional: Check if the category has items and handle accordingly
# For now, we'll allow deletion even if items are associated with it
return category.remove(db, id=id)

View File

@ -0,0 +1,32 @@
from typing import Any, Dict
from fastapi import APIRouter, Depends
from sqlalchemy import text
from sqlalchemy.orm import Session
from app.db.utils import get_db
router = APIRouter()
@router.get("/", response_model=Dict[str, Any])
async def health_check(db: Session = Depends(get_db)) -> Any:
"""
Health check endpoint to verify the API and database are running.
"""
health_status = {
"status": "healthy",
"api": "ok",
"database": "ok",
}
# Check database connection
try:
# Execute a simple query to check if the database is responsive
db.execute(text("SELECT 1"))
except Exception as e:
health_status["status"] = "unhealthy"
health_status["database"] = "error"
health_status["database_error"] = str(e)
return health_status

View File

@ -0,0 +1,198 @@
from typing import Any, List, Optional
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.api.deps import get_current_user
from app.crud import category, item
from app.db.utils import get_db
from app.models.user import User
from app.schemas.item import Item, ItemCreate, ItemUpdate
router = APIRouter()
@router.get("/", response_model=List[Item])
def read_items(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
category_id: Optional[int] = None,
low_stock: Optional[bool] = False,
name: Optional[str] = None,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Retrieve items.
- Optional filter by category_id
- Optional filter for low stock items
- Optional search by name
"""
if low_stock:
items = item.get_low_stock_items(db, skip=skip, limit=limit)
elif category_id:
items = item.get_by_category_id(
db, category_id=category_id, skip=skip, limit=limit
)
elif name:
items = item.get_by_name(db, name=name)
else:
items = item.get_multi(db, skip=skip, limit=limit)
return items
@router.post("/", response_model=Item)
def create_item(
*,
db: Session = Depends(get_db),
item_in: ItemCreate,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Create new item.
"""
# Check if SKU already exists
if item.get_by_sku(db, sku=item_in.sku):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="An item with this SKU already exists.",
)
# Check if category exists if provided
if item_in.category_id is not None:
db_category = category.get(db, id=item_in.category_id)
if not db_category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found",
)
# Create item with current user as owner
return item.create_with_owner(db, obj_in=item_in, owner_id=current_user.id)
@router.get("/{id}", response_model=Item)
def read_item(
*,
db: Session = Depends(get_db),
id: int,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get item by ID.
"""
db_item = item.get(db, id=id)
if not db_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
return db_item
@router.put("/{id}", response_model=Item)
def update_item(
*,
db: Session = Depends(get_db),
id: int,
item_in: ItemUpdate,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Update an item.
"""
db_item = item.get(db, id=id)
if not db_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
# Check if SKU is being updated and ensure it doesn't conflict
if item_in.sku and item_in.sku != db_item.sku:
existing_item = item.get_by_sku(db, sku=item_in.sku)
if existing_item:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="An item with this SKU already exists.",
)
# Check if category exists if provided
if item_in.category_id is not None:
db_category = category.get(db, id=item_in.category_id)
if not db_category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found",
)
# Only superusers or the owner can update items
if not current_user.is_superuser and db_item.owner_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
return item.update(db, db_obj=db_item, obj_in=item_in)
@router.delete("/{id}", response_model=Item)
def delete_item(
*,
db: Session = Depends(get_db),
id: int,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Delete an item.
"""
db_item = item.get(db, id=id)
if not db_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
# Only superusers or the owner can delete items
if not current_user.is_superuser and db_item.owner_id != current_user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
return item.remove(db, id=id)
@router.get("/by-sku/{sku}", response_model=Item)
def read_item_by_sku(
*,
db: Session = Depends(get_db),
sku: str,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get item by SKU.
"""
db_item = item.get_by_sku(db, sku=sku)
if not db_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
return db_item
@router.get("/low-stock/", response_model=List[Item])
def read_low_stock_items(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Retrieve items with stock below reorder level.
"""
items = item.get_low_stock_items(db, skip=skip, limit=limit)
return items

View File

@ -0,0 +1,168 @@
from datetime import date, datetime
from typing import Any, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from app.api.deps import get_current_user
from app.crud import category, item, transaction
from app.db.utils import get_db
from app.models.user import User
from app.schemas.report import (
CategorySummaryItem,
CategorySummaryReport,
InventoryValueItem,
InventoryValueReport,
LowStockItem,
LowStockReport,
TransactionSummaryItem,
TransactionSummaryReport,
)
router = APIRouter()
@router.get("/inventory-value", response_model=InventoryValueReport)
def get_inventory_value(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Generate inventory value report.
"""
items_db = item.get_multi(db)
inventory_items = []
total_value = 0.0
for item_db in items_db:
item_value = item_db.price * item_db.quantity
total_value += item_value
inventory_items.append(
InventoryValueItem(
id=item_db.id,
name=item_db.name,
sku=item_db.sku,
quantity=item_db.quantity,
price=item_db.price,
total_value=item_value
)
)
return InventoryValueReport(
total_inventory_value=total_value,
items=inventory_items,
report_date=datetime.utcnow()
)
@router.get("/category-summary", response_model=CategorySummaryReport)
def get_category_summary(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Generate category summary report.
"""
# Get all categories
categories_db = category.get_multi(db)
summary_items = []
for cat in categories_db:
# Get items in this category
items_in_category = item.get_by_category_id(db, category_id=cat.id)
# Calculate totals
item_count = len(items_in_category)
total_value = sum(i.price * i.quantity for i in items_in_category)
avg_price = sum(i.price for i in items_in_category) / item_count if item_count > 0 else 0
summary_items.append(
CategorySummaryItem(
category_id=cat.id,
category_name=cat.name,
item_count=item_count,
total_value=total_value,
average_price=avg_price
)
)
return CategorySummaryReport(
categories=summary_items,
report_date=datetime.utcnow()
)
@router.get("/low-stock", response_model=LowStockReport)
def get_low_stock_report(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Generate low stock report.
"""
# Get items with stock below reorder level
low_stock_items_db = item.get_low_stock_items(db)
low_stock_items = []
for item_db in low_stock_items_db:
low_stock_items.append(
LowStockItem(
id=item_db.id,
name=item_db.name,
sku=item_db.sku,
quantity=item_db.quantity,
reorder_level=item_db.reorder_level,
category_name=item_db.category.name if item_db.category else None
)
)
return LowStockReport(
low_stock_items=low_stock_items,
report_date=datetime.utcnow()
)
@router.get("/transaction-summary", response_model=TransactionSummaryReport)
def get_transaction_summary(
start_date: date = Query(..., description="Start date for the report period"),
end_date: Optional[date] = Query(None, description="End date for the report period"),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Generate transaction summary report.
"""
if end_date is None:
end_date = date.today()
if start_date > end_date:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Start date must be before end date",
)
# Get transaction summary by type
summary_data = transaction.get_summary_by_type(
db, start_date=start_date, end_date=end_date
)
summary_items = [
TransactionSummaryItem(
transaction_type=item["transaction_type"],
count=item["count"],
total_quantity=item["total_quantity"],
total_value=item["total_value"]
)
for item in summary_data
]
return TransactionSummaryReport(
start_date=start_date,
end_date=end_date,
transactions=summary_items,
report_date=datetime.utcnow()
)

View File

@ -0,0 +1,105 @@
from datetime import date
from typing import Any, List, Optional
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.api.deps import get_current_user
from app.crud import item, transaction
from app.db.utils import get_db
from app.models.transaction import TransactionType
from app.models.user import User
from app.schemas.transaction import Transaction, TransactionCreate
router = APIRouter()
@router.get("/", response_model=List[Transaction])
def read_transactions(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
item_id: Optional[int] = None,
start_date: Optional[date] = None,
end_date: Optional[date] = None,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Retrieve transactions.
- Optional filter by item_id
- Optional filter by date range
"""
if item_id:
return transaction.get_by_item_id(db, item_id=item_id, skip=skip, limit=limit)
elif start_date:
return transaction.get_by_date_range(
db, start_date=start_date, end_date=end_date, skip=skip, limit=limit
)
else:
return transaction.get_multi(db, skip=skip, limit=limit)
@router.post("/", response_model=Transaction)
def create_transaction(
*,
db: Session = Depends(get_db),
transaction_in: TransactionCreate,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Create new transaction and update item quantity.
"""
# Check if item exists
db_item = item.get(db, id=transaction_in.item_id)
if not db_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
# Create the transaction
db_transaction = transaction.create_with_item_and_user(
db, obj_in=transaction_in, recorded_by_id=current_user.id
)
# Update item quantity based on transaction type
if transaction_in.transaction_type == TransactionType.PURCHASE:
db_item.quantity += transaction_in.quantity
elif transaction_in.transaction_type == TransactionType.SALE:
if db_item.quantity < transaction_in.quantity:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Not enough items in stock",
)
db_item.quantity -= transaction_in.quantity
elif transaction_in.transaction_type == TransactionType.ADJUSTMENT:
db_item.quantity += transaction_in.quantity # Can be negative for decrease
elif transaction_in.transaction_type == TransactionType.RETURN:
db_item.quantity += transaction_in.quantity
# Save the updated item
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_transaction
@router.get("/{id}", response_model=Transaction)
def read_transaction(
*,
db: Session = Depends(get_db),
id: int,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get transaction by ID.
"""
db_transaction = transaction.get(db, id=id)
if not db_transaction:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Transaction not found",
)
return db_transaction

View File

@ -0,0 +1,131 @@
from typing import Any, List
from fastapi import APIRouter, Body, Depends, HTTPException, status
from fastapi.encoders import jsonable_encoder
from pydantic import EmailStr
from sqlalchemy.orm import Session
from app.api.deps import get_current_active_superuser, get_current_user
from app.crud import user
from app.db.utils import get_db
from app.models.user import User
from app.schemas.user import User as UserSchema
from app.schemas.user import UserCreate, UserUpdate
router = APIRouter()
@router.get("/", response_model=List[UserSchema])
def read_users(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(get_current_active_superuser),
) -> Any:
"""
Retrieve users.
"""
users = user.get_multi(db, skip=skip, limit=limit)
return users
@router.post("/", response_model=UserSchema)
def create_user(
*,
db: Session = Depends(get_db),
user_in: UserCreate,
current_user: User = Depends(get_current_active_superuser),
) -> Any:
"""
Create new user.
"""
db_user = user.get_by_email(db, email=user_in.email)
if db_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="The user with this email already exists in the system.",
)
new_user = user.create(db, obj_in=user_in)
return new_user
@router.get("/me", response_model=UserSchema)
def read_user_me(
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get current user.
"""
return current_user
@router.put("/me", response_model=UserSchema)
def update_user_me(
*,
db: Session = Depends(get_db),
password: str = Body(None),
full_name: str = Body(None),
email: EmailStr = Body(None),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Update own user.
"""
current_user_data = jsonable_encoder(current_user)
user_in = UserUpdate(**current_user_data)
if password is not None:
user_in.password = password
if full_name is not None:
user_in.full_name = full_name
if email is not None:
user_in.email = email
updated_user = user.update(db, db_obj=current_user, obj_in=user_in)
return updated_user
@router.get("/{user_id}", response_model=UserSchema)
def read_user_by_id(
user_id: int,
current_user: User = Depends(get_current_user),
db: Session = Depends(get_db),
) -> Any:
"""
Get a specific user by id.
"""
db_user = user.get(db, id=user_id)
if not db_user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
# Regular users can only view their own details
if not current_user.is_superuser and current_user.id != user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
return db_user
@router.put("/{user_id}", response_model=UserSchema)
def update_user(
*,
db: Session = Depends(get_db),
user_id: int,
user_in: UserUpdate,
current_user: User = Depends(get_current_active_superuser),
) -> Any:
"""
Update a user.
"""
db_user = user.get(db, id=user_id)
if not db_user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
updated_user = user.update(db, db_obj=db_user, obj_in=user_in)
return updated_user

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

@ -0,0 +1,43 @@
from pathlib import Path
from typing import List, Union
from pydantic import AnyHttpUrl, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=True,
)
# API settings
API_V1_STR: str = "/api/v1"
PROJECT_NAME: str = "Small Business Inventory Management System"
# CORS settings
BACKEND_CORS_ORIGINS: List[AnyHttpUrl] = []
@field_validator("BACKEND_CORS_ORIGINS", mode="before")
def assemble_cors_origins(cls, v: Union[str, List[str]]) -> Union[List[str], str]:
if isinstance(v, str) and not v.startswith("["):
return [i.strip() for i in v.split(",")]
elif isinstance(v, (list, str)):
return v
raise ValueError(v)
# Database settings
DB_DIR: Path = Path("/app") / "storage" / "db"
SQLALCHEMY_DATABASE_URL: str = f"sqlite:///{DB_DIR}/db.sqlite"
# Security settings
SECRET_KEY: str = "secret_key_for_development_only_please_change_in_production"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8 # 8 days
ALGORITHM: str = "HS256"
settings = Settings()
# Create the DB directory if it doesn't exist
settings.DB_DIR.mkdir(parents=True, exist_ok=True)

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

@ -0,0 +1,40 @@
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=settings.ALGORITHM)
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)

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

@ -0,0 +1,5 @@
# Import CRUD operations to make them accessible
from app.crud.category import category
from app.crud.item import item
from app.crud.transaction import transaction
from app.crud.user import user

83
app/crud/base.py Normal file
View File

@ -0,0 +1,83 @@
from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union
from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel
from sqlalchemy.orm import Session
from app.models.base import Base
ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
"""
Base class for CRUD operations.
"""
def __init__(self, model: Type[ModelType]):
"""
CRUD object with default methods to Create, Read, Update, Delete (CRUD).
**Parameters**
* `model`: A SQLAlchemy model class
"""
self.model = model
def get(self, db: Session, id: Any) -> Optional[ModelType]:
"""
Get a single record by ID.
"""
return db.query(self.model).filter(self.model.id == id).first()
def get_multi(
self, db: Session, *, skip: int = 0, limit: int = 100
) -> List[ModelType]:
"""
Get multiple records.
"""
return db.query(self.model).offset(skip).limit(limit).all()
def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:
"""
Create a new record.
"""
obj_in_data = jsonable_encoder(obj_in)
db_obj = self.model(**obj_in_data)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def update(
self,
db: Session,
*,
db_obj: ModelType,
obj_in: Union[UpdateSchemaType, Dict[str, Any]]
) -> ModelType:
"""
Update a record.
"""
obj_data = jsonable_encoder(db_obj)
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.model_dump(exclude_unset=True)
for field in obj_data:
if field in update_data:
setattr(db_obj, field, update_data[field])
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def remove(self, db: Session, *, id: int) -> ModelType:
"""
Delete a record.
"""
obj = db.query(self.model).get(id)
db.delete(obj)
db.commit()
return obj

35
app/crud/category.py Normal file
View File

@ -0,0 +1,35 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.crud.base import CRUDBase
from app.models.category import Category
from app.schemas.category import CategoryCreate, CategoryUpdate
class CRUDCategory(CRUDBase[Category, CategoryCreate, CategoryUpdate]):
"""
CRUD operations for Category model.
"""
def get_by_name(self, db: Session, *, name: str) -> Optional[Category]:
"""
Get a category by name.
"""
return db.query(Category).filter(Category.name == name).first()
def get_multi_with_items(
self, db: Session, *, skip: int = 0, limit: int = 100
) -> List[Category]:
"""
Get multiple categories with items.
"""
return (
db.query(Category)
.offset(skip)
.limit(limit)
.all()
)
# Create a singleton instance
category = CRUDCategory(Category)

69
app/crud/item.py Normal file
View File

@ -0,0 +1,69 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.crud.base import CRUDBase
from app.models.item import Item
from app.schemas.item import ItemCreate, ItemUpdate
class CRUDItem(CRUDBase[Item, ItemCreate, ItemUpdate]):
"""
CRUD operations for Item model.
"""
def get_by_sku(self, db: Session, *, sku: str) -> Optional[Item]:
"""
Get an item by SKU.
"""
return db.query(Item).filter(Item.sku == sku).first()
def get_by_name(self, db: Session, *, name: str) -> List[Item]:
"""
Get items by name (can return multiple items).
"""
return db.query(Item).filter(Item.name.ilike(f"%{name}%")).all()
def get_by_category_id(
self, db: Session, *, category_id: int, skip: int = 0, limit: int = 100
) -> List[Item]:
"""
Get items by category ID.
"""
return (
db.query(Item)
.filter(Item.category_id == category_id)
.offset(skip)
.limit(limit)
.all()
)
def get_low_stock_items(
self, db: Session, *, skip: int = 0, limit: int = 100
) -> List[Item]:
"""
Get items with stock below reorder level.
"""
return (
db.query(Item)
.filter(Item.quantity <= Item.reorder_level)
.offset(skip)
.limit(limit)
.all()
)
def create_with_owner(
self, db: Session, *, obj_in: ItemCreate, owner_id: int
) -> Item:
"""
Create a new item with owner.
"""
obj_in_data = obj_in.model_dump()
db_obj = Item(**obj_in_data, owner_id=owner_id)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
# Create a singleton instance
item = CRUDItem(Item)

111
app/crud/transaction.py Normal file
View File

@ -0,0 +1,111 @@
from datetime import date, datetime
from typing import List
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.crud.base import CRUDBase
from app.models.transaction import Transaction
from app.schemas.transaction import TransactionCreate, TransactionUpdate
class CRUDTransaction(CRUDBase[Transaction, TransactionCreate, TransactionUpdate]):
"""
CRUD operations for Transaction model.
"""
def create_with_item_and_user(
self, db: Session, *, obj_in: TransactionCreate, recorded_by_id: int
) -> Transaction:
"""
Create a new transaction record with item and user.
"""
obj_in_data = obj_in.model_dump()
db_obj = Transaction(**obj_in_data, recorded_by_id=recorded_by_id)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def get_by_item_id(
self, db: Session, *, item_id: int, skip: int = 0, limit: int = 100
) -> List[Transaction]:
"""
Get transactions by item ID.
"""
return (
db.query(Transaction)
.filter(Transaction.item_id == item_id)
.order_by(Transaction.transaction_date.desc())
.offset(skip)
.limit(limit)
.all()
)
def get_by_date_range(
self,
db: Session,
*,
start_date: date,
end_date: date = None,
skip: int = 0,
limit: int = 100
) -> List[Transaction]:
"""
Get transactions within a date range.
"""
if end_date is None:
end_date = date.today()
# Convert dates to datetime for proper comparison
start_datetime = datetime.combine(start_date, datetime.min.time())
end_datetime = datetime.combine(end_date, datetime.max.time())
return (
db.query(Transaction)
.filter(Transaction.transaction_date >= start_datetime)
.filter(Transaction.transaction_date <= end_datetime)
.order_by(Transaction.transaction_date.desc())
.offset(skip)
.limit(limit)
.all()
)
def get_summary_by_type(
self, db: Session, *, start_date: date, end_date: date = None
) -> List[dict]:
"""
Get transaction summary aggregated by transaction type.
"""
if end_date is None:
end_date = date.today()
# Convert dates to datetime for proper comparison
start_datetime = datetime.combine(start_date, datetime.min.time())
end_datetime = datetime.combine(end_date, datetime.max.time())
result = (
db.query(
Transaction.transaction_type,
func.count().label("count"),
func.sum(Transaction.quantity).label("total_quantity"),
func.sum(Transaction.unit_price * Transaction.quantity).label("total_value")
)
.filter(Transaction.transaction_date >= start_datetime)
.filter(Transaction.transaction_date <= end_datetime)
.group_by(Transaction.transaction_type)
.all()
)
return [
{
"transaction_type": r[0],
"count": r[1],
"total_quantity": r[2],
"total_value": r[3]
}
for r in result
]
# Create a singleton instance
transaction = CRUDTransaction(Transaction)

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

@ -0,0 +1,80 @@
from typing import Any, Dict, Optional, Union
from sqlalchemy.orm import Session
from app.core.security import get_password_hash, verify_password
from app.crud.base import CRUDBase
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
"""
CRUD operations for User model.
"""
def get_by_email(self, db: Session, *, email: str) -> Optional[User]:
"""
Get a user by email.
"""
return db.query(User).filter(User.email == email).first()
def create(self, 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_superuser=obj_in.is_superuser,
is_active=obj_in.is_active,
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def update(
self, 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.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
return super().update(db, db_obj=db_obj, obj_in=update_data)
def authenticate(self, db: Session, *, email: str, password: str) -> Optional[User]:
"""
Authenticate a user.
"""
user = self.get_by_email(db, email=email)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user
def is_active(self, user: User) -> bool:
"""
Check if a user is active.
"""
return user.is_active
def is_superuser(self, user: User) -> bool:
"""
Check if a user is a superuser.
"""
return user.is_superuser
# Create a singleton instance
user = CRUDUser(User)

17
app/db/init_db.py Normal file
View File

@ -0,0 +1,17 @@
import logging
from app.db.session import Base, engine
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def init_db() -> None:
"""
Initialize the database.
Creates all tables defined in the models.
"""
# Create tables
logger.info("Creating database tables")
Base.metadata.create_all(bind=engine)
logger.info("Database tables created")

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

@ -0,0 +1,17 @@
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
# Create the SQLAlchemy engine
engine = create_engine(
settings.SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False} # Only needed for SQLite
)
# Create a SessionLocal class for database sessions
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# Create a Base class for declarative models
Base = declarative_base()

16
app/db/utils.py Normal file
View File

@ -0,0 +1,16 @@
from typing import Generator
from app.db.session import SessionLocal
# Dependency to get DB session
def get_db() -> Generator:
"""
Dependency for getting a database session.
This function yields a SQLAlchemy session and ensures it's closed after use.
"""
db = SessionLocal()
try:
yield db
finally:
db.close()

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

@ -0,0 +1,5 @@
# Import models to ensure they are registered with SQLAlchemy
from app.models.category import Category
from app.models.item import Item
from app.models.transaction import Transaction, TransactionType
from app.models.user import User

26
app/models/base.py Normal file
View File

@ -0,0 +1,26 @@
from datetime import datetime
from typing import Any
from sqlalchemy import Column, DateTime
from sqlalchemy.ext.declarative import as_declarative, declared_attr
@as_declarative()
class Base:
"""
Base class for all models.
Provides common columns and methods.
"""
id: Any
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(
DateTime,
default=datetime.utcnow,
onupdate=datetime.utcnow,
nullable=False
)
# Generate __tablename__ automatically based on class name
@declared_attr
def __tablename__(cls) -> str:
return cls.__name__.lower()

12
app/models/category.py Normal file
View File

@ -0,0 +1,12 @@
from sqlalchemy import Column, Integer, String, Text
from app.models.base import Base
class Category(Base):
"""
Category model for grouping inventory items.
"""
id = Column(Integer, primary_key=True, index=True)
name = Column(String, unique=True, index=True, nullable=False)
description = Column(Text, nullable=True)

25
app/models/item.py Normal file
View File

@ -0,0 +1,25 @@
from sqlalchemy import Column, Float, ForeignKey, Integer, String, Text
from sqlalchemy.orm import relationship
from app.models.base import Base
class Item(Base):
"""
Inventory item model.
"""
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True, nullable=False)
sku = Column(String, index=True, unique=True, nullable=False)
description = Column(Text, nullable=True)
price = Column(Float, nullable=False)
quantity = Column(Integer, default=0, nullable=False)
reorder_level = Column(Integer, default=10, nullable=False)
# Foreign key to Category
category_id = Column(Integer, ForeignKey("category.id"), nullable=True)
category = relationship("Category", backref="items")
# Foreign key to User (who created the item)
owner_id = Column(Integer, ForeignKey("user.id"), nullable=True)
owner = relationship("User", backref="items")

38
app/models/transaction.py Normal file
View File

@ -0,0 +1,38 @@
from datetime import datetime
from enum import Enum as PyEnum
from sqlalchemy import Column, DateTime, Enum, Float, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from app.models.base import Base
class TransactionType(str, PyEnum):
"""
Enum for transaction types.
"""
PURCHASE = "purchase"
SALE = "sale"
ADJUSTMENT = "adjustment"
RETURN = "return"
class Transaction(Base):
"""
Transaction model for tracking inventory changes.
"""
id = Column(Integer, primary_key=True, index=True)
transaction_type = Column(Enum(TransactionType), nullable=False)
quantity = Column(Integer, nullable=False)
unit_price = Column(Float, nullable=True)
reference_number = Column(String, nullable=True)
notes = Column(String, nullable=True)
transaction_date = Column(DateTime, default=datetime.utcnow, nullable=False)
# Foreign key to Item
item_id = Column(Integer, ForeignKey("item.id"), nullable=False)
item = relationship("Item", backref="transactions")
# Foreign key to User (who recorded the transaction)
recorded_by_id = Column(Integer, ForeignKey("user.id"), nullable=True)
recorded_by = relationship("User", backref="transactions")

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

@ -0,0 +1,15 @@
from sqlalchemy import Boolean, Column, Integer, String
from app.models.base import Base
class User(Base):
"""
User model for authentication and authorization.
"""
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, nullable=False)
is_superuser = Column(Boolean, default=False, nullable=False)

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

@ -0,0 +1,17 @@
# Import schemas to make them accessible from app.schemas
from app.schemas.category import Category, CategoryCreate, CategoryInDB, CategoryUpdate
from app.schemas.item import Item, ItemCreate, ItemInDB, ItemUpdate
from app.schemas.report import (
CategorySummaryReport,
InventoryValueReport,
LowStockReport,
TransactionSummaryReport,
)
from app.schemas.token import Token, TokenPayload
from app.schemas.transaction import (
Transaction,
TransactionCreate,
TransactionInDB,
TransactionUpdate,
)
from app.schemas.user import User, UserCreate, UserInDB, UserUpdate

38
app/schemas/category.py Normal file
View File

@ -0,0 +1,38 @@
from typing import Optional
from pydantic import BaseModel
# Shared properties
class CategoryBase(BaseModel):
name: Optional[str] = None
description: Optional[str] = None
# Properties to receive on category creation
class CategoryCreate(CategoryBase):
name: str
# Properties to receive on category update
class CategoryUpdate(CategoryBase):
pass
# Properties shared by models in DB
class CategoryInDBBase(CategoryBase):
id: int
name: str
class Config:
from_attributes = True
# Properties to return to client
class Category(CategoryInDBBase):
pass
# Properties stored in DB
class CategoryInDB(CategoryInDBBase):
pass

48
app/schemas/item.py Normal file
View File

@ -0,0 +1,48 @@
from typing import Optional
from pydantic import BaseModel
# Shared properties
class ItemBase(BaseModel):
name: Optional[str] = None
sku: Optional[str] = None
description: Optional[str] = None
price: Optional[float] = None
quantity: Optional[int] = 0
reorder_level: Optional[int] = 10
category_id: Optional[int] = None
# Properties to receive on item creation
class ItemCreate(ItemBase):
name: str
sku: str
price: float
# Properties to receive on item update
class ItemUpdate(ItemBase):
pass
# Properties shared by models in DB
class ItemInDBBase(ItemBase):
id: int
name: str
sku: str
price: float
owner_id: Optional[int] = None
class Config:
from_attributes = True
# Properties to return to client
class Item(ItemInDBBase):
pass
# Properties stored in DB
class ItemInDB(ItemInDBBase):
pass

62
app/schemas/report.py Normal file
View File

@ -0,0 +1,62 @@
from datetime import date, datetime
from typing import List, Optional
from pydantic import BaseModel
from app.models.transaction import TransactionType
class InventoryValueItem(BaseModel):
id: int
name: str
sku: str
quantity: int
price: float
total_value: float
class InventoryValueReport(BaseModel):
total_inventory_value: float
items: List[InventoryValueItem]
report_date: datetime = datetime.utcnow()
class CategorySummaryItem(BaseModel):
category_id: int
category_name: str
item_count: int
total_value: float
average_price: float
class CategorySummaryReport(BaseModel):
categories: List[CategorySummaryItem]
report_date: datetime = datetime.utcnow()
class LowStockItem(BaseModel):
id: int
name: str
sku: str
quantity: int
reorder_level: int
category_name: Optional[str] = None
class LowStockReport(BaseModel):
low_stock_items: List[LowStockItem]
report_date: datetime = datetime.utcnow()
class TransactionSummaryItem(BaseModel):
transaction_type: TransactionType
count: int
total_quantity: int
total_value: Optional[float] = None
class TransactionSummaryReport(BaseModel):
start_date: date
end_date: date
transactions: List[TransactionSummaryItem]
report_date: datetime = datetime.utcnow()

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

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

View File

@ -0,0 +1,53 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel
from app.models.transaction import TransactionType
# Shared properties
class TransactionBase(BaseModel):
transaction_type: Optional[TransactionType] = None
quantity: Optional[int] = None
unit_price: Optional[float] = None
reference_number: Optional[str] = None
notes: Optional[str] = None
transaction_date: Optional[datetime] = None
item_id: Optional[int] = None
# Properties to receive on transaction creation
class TransactionCreate(TransactionBase):
transaction_type: TransactionType
quantity: int
item_id: int
transaction_date: Optional[datetime] = datetime.utcnow()
# Properties to receive on transaction update
class TransactionUpdate(TransactionBase):
pass
# Properties shared by models in DB
class TransactionInDBBase(TransactionBase):
id: int
transaction_type: TransactionType
quantity: int
item_id: int
transaction_date: datetime
recorded_by_id: Optional[int] = None
class Config:
from_attributes = True
# Properties to return to client
class Transaction(TransactionInDBBase):
pass
# Properties stored in DB
class TransactionInDB(TransactionInDBBase):
pass

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

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

38
main.py Normal file
View File

@ -0,0 +1,38 @@
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 FastAPI app
app = FastAPI(
title=settings.PROJECT_NAME,
openapi_url="/openapi.json",
)
# Configure CORS
if settings.BACKEND_CORS_ORIGINS:
app.add_middleware(
CORSMiddleware,
allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include API router
app.include_router(api_router, prefix=settings.API_V1_STR)
# Health check endpoint
@app.get("/health", response_model=None)
async def health_check():
"""
Health check endpoint to verify the API is running.
"""
return JSONResponse(content={"status": "healthy"})
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

1
migrations/__init__.py Normal file
View File

@ -0,0 +1 @@
# This file is here to make the migrations directory a Python package.

86
migrations/env.py Normal file
View File

@ -0,0 +1,86 @@
import os
import sys
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
# Add the project root directory to the Python path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Import the SQLAlchemy metadata object
from app.db.session import Base
from app.models import * # Import all models for Alembic to detect them
# 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, # Key configuration 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 @@
# This file is here to make the versions directory a Python package.

View File

@ -0,0 +1,114 @@
"""Initial migration
Revision ID: 001
Revises:
Create Date: 2023-07-20
"""
import sqlalchemy as sa
from alembic import op
# 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('full_name', sa.String(), nullable=True),
sa.Column('hashed_password', sa.String(), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=False, default=True),
sa.Column('is_superuser', sa.Boolean(), nullable=False, default=False),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
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 category table
op.create_table(
'category',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_category_id'), 'category', ['id'], unique=False)
op.create_index(op.f('ix_category_name'), 'category', ['name'], unique=True)
# Create item table
op.create_table(
'item',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('sku', sa.String(), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('price', sa.Float(), nullable=False),
sa.Column('quantity', sa.Integer(), nullable=False, default=0),
sa.Column('reorder_level', sa.Integer(), nullable=False, default=10),
sa.Column('category_id', sa.Integer(), nullable=True),
sa.Column('owner_id', sa.Integer(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['category_id'], ['category.id'], ),
sa.ForeignKeyConstraint(['owner_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_item_id'), 'item', ['id'], unique=False)
op.create_index(op.f('ix_item_name'), 'item', ['name'], unique=False)
op.create_index(op.f('ix_item_sku'), 'item', ['sku'], unique=True)
# Create transaction table
op.create_table(
'transaction',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column(
'transaction_type',
sa.Enum('purchase', 'sale', 'adjustment', 'return', name='transactiontype'),
nullable=False
),
sa.Column('quantity', sa.Integer(), nullable=False),
sa.Column('unit_price', sa.Float(), nullable=True),
sa.Column('reference_number', sa.String(), nullable=True),
sa.Column('notes', sa.String(), nullable=True),
sa.Column('transaction_date', sa.DateTime(), nullable=False),
sa.Column('item_id', sa.Integer(), nullable=False),
sa.Column('recorded_by_id', sa.Integer(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['item_id'], ['item.id'], ),
sa.ForeignKeyConstraint(['recorded_by_id'], ['user.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_transaction_id'), 'transaction', ['id'], unique=False)
def downgrade():
# Drop transaction table
op.drop_index(op.f('ix_transaction_id'), table_name='transaction')
op.drop_table('transaction')
# Drop item table
op.drop_index(op.f('ix_item_sku'), table_name='item')
op.drop_index(op.f('ix_item_name'), table_name='item')
op.drop_index(op.f('ix_item_id'), table_name='item')
op.drop_table('item')
# Drop category table
op.drop_index(op.f('ix_category_name'), table_name='category')
op.drop_index(op.f('ix_category_id'), table_name='category')
op.drop_table('category')
# 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')

12
pyproject.toml Normal file
View File

@ -0,0 +1,12 @@
[tool.ruff]
line-length = 120
indent-width = 4
target-version = "py38"
[tool.ruff.lint]
select = ["E", "F", "B", "I"]
ignore = ["B008", "F403", "F401"] # Allow Depends() calls in function defaults, star imports and unused imports
[tool.ruff.format]
quote-style = "double"
indent-style = "space"

12
requirements.txt Normal file
View File

@ -0,0 +1,12 @@
fastapi>=0.95.0
uvicorn>=0.22.0
sqlalchemy>=2.0.0
alembic>=1.11.0
pydantic>=2.0.0
pydantic-settings>=2.0.0
python-jose[cryptography]>=3.3.0
passlib[bcrypt]>=1.7.4
python-multipart>=0.0.6
ruff>=0.0.262
python-dotenv>=1.0.0
email-validator>=2.0.0