Implement Small Business Inventory Management System

- Created FastAPI application with SQLite database
- Implemented models for inventory items, categories, suppliers, and transactions
- Added authentication system with JWT tokens
- Implemented CRUD operations for all models
- Set up Alembic for database migrations
- Added comprehensive API documentation
- Configured Ruff for code linting
This commit is contained in:
Automated Action 2025-06-06 20:27:41 +00:00
parent b869cbc4c7
commit 1d312e5ff6
40 changed files with 2324 additions and 2 deletions

150
README.md
View File

@ -1,3 +1,149 @@
# FastAPI Application
# Small Business Inventory Management System
This is a FastAPI application bootstrapped by BackendIM, the AI-powered backend generation platform.
A robust API-based inventory management system built for small businesses using FastAPI and SQLite. This system helps businesses track inventory items, manage suppliers, categorize products, and record stock movements.
## Features
- **Item Management**: Create, update, delete, and search inventory items
- **Category Management**: Organize items by categories
- **Supplier Management**: Track suppliers for inventory items
- **Stock Transactions**: Record stock-in and stock-out transactions
- **User Authentication**: Secure API access with JWT authentication
- **Low Stock Alerts**: Identify items with stock levels below reorder points
- **Reporting**: Generate transaction reports for specific time periods
## Technology Stack
- **Backend**: Python 3.8+ with FastAPI
- **Database**: SQLite with SQLAlchemy ORM
- **Authentication**: JWT tokens with OAuth2
- **API Documentation**: Swagger UI and ReDoc via OpenAPI
- **Migrations**: Alembic for database schema versioning
## API Documentation
API documentation is available at:
- Swagger UI: `/docs`
- ReDoc: `/redoc`
## Environment Variables
The application uses the following environment variables:
| Variable | Description | Default |
|----------|-------------|---------|
| SECRET_KEY | JWT secret key | super-secret-key-for-development-only |
| ACCESS_TOKEN_EXPIRE_MINUTES | JWT token expiration time in minutes | 60 |
## Installation and Setup
### Prerequisites
- Python 3.8 or higher
- pip (Python package manager)
### Installation Steps
1. Clone the repository:
```
git clone <repository-url>
cd small-business-inventory-management-system
```
2. Install dependencies:
```
pip install -r requirements.txt
```
3. Run database migrations:
```
alembic upgrade head
```
4. Start the application:
```
uvicorn main:app --reload
```
The API will be available at http://localhost:8000
## API Endpoints
### Authentication
- `POST /api/v1/auth/login` - Obtain JWT token
- `POST /api/v1/auth/register` - Register new user
- `GET /api/v1/auth/me` - Get current user info
### Items
- `GET /api/v1/items` - List all items (with optional filtering)
- `POST /api/v1/items` - Create new item
- `GET /api/v1/items/{item_id}` - Get item details
- `PUT /api/v1/items/{item_id}` - Update item
- `DELETE /api/v1/items/{item_id}` - Delete item
- `GET /api/v1/items/search` - Search items by name, SKU, or barcode
### Categories
- `GET /api/v1/categories` - List all categories
- `POST /api/v1/categories` - Create new category
- `GET /api/v1/categories/{category_id}` - Get category details
- `PUT /api/v1/categories/{category_id}` - Update category
- `DELETE /api/v1/categories/{category_id}` - Delete category
### Suppliers
- `GET /api/v1/suppliers` - List all suppliers
- `POST /api/v1/suppliers` - Create new supplier
- `GET /api/v1/suppliers/{supplier_id}` - Get supplier details
- `PUT /api/v1/suppliers/{supplier_id}` - Update supplier
- `DELETE /api/v1/suppliers/{supplier_id}` - Delete supplier
### Transactions
- `GET /api/v1/transactions` - List all transactions (with optional filtering)
- `POST /api/v1/transactions` - Create new transaction (stock in/out)
- `GET /api/v1/transactions/{transaction_id}` - Get transaction details
- `PUT /api/v1/transactions/{transaction_id}` - Update transaction metadata
- `GET /api/v1/transactions/by-item/{item_id}` - Get item transactions
- `GET /api/v1/transactions/report/daily` - Get daily transaction report
## Data Models
### User
- Username, email, password, full name
- Active status and superuser privileges
### Item
- Name, description, SKU, barcode
- Quantity, unit price, reorder level
- Associated category and supplier
### Category
- Name and description
### Supplier
- Name, contact info (name, email, phone, address)
### Transaction
- Associated item and user
- Quantity, unit price, total price
- Transaction type (stock in/out)
- Reference, notes, timestamp
## Development
### Adding Database Migrations
When you make changes to the database models, generate new migrations with:
```
alembic revision --autogenerate -m "Description of changes"
```
Then apply the migrations:
```
alembic upgrade head
```

85
alembic.ini Normal file
View File

@ -0,0 +1,85 @@
# A generic, single database configuration.
[alembic]
# path to migration scripts
script_location = migrations
# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s
# timezone to use when rendering the date
# within the migration file as well as the filename.
# string value is passed to dateutil.tz.gettz()
# leave blank for localtime
# timezone =
# max length of characters to apply to the
# "slug" field
# truncate_slug_length = 40
# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false
# set to 'true' to allow .pyc and .pyo files without
# a source .py file to be detected as revisions in the
# versions/ directory
# sourceless = false
# version location specification; this defaults
# to migrations/versions. When using multiple version
# directories, initial revisions must be specified with --version-path
# version_locations = %(here)s/bar %(here)s/bat migrations/versions
# the output encoding used when revision files
# are written from script.py.mako
# output_encoding = utf-8
# SQLite URL - using absolute path
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/api/deps.py Normal file
View File

@ -0,0 +1,3 @@
from app.db.deps import get_current_user, get_db
__all__ = ["get_db", "get_current_user"]

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

@ -0,0 +1,11 @@
from fastapi import APIRouter
from app.api.v1.endpoints import auth, categories, items, suppliers, transactions
api_router = APIRouter()
api_router.include_router(auth.router, prefix="/auth", tags=["authentication"])
api_router.include_router(items.router, prefix="/items", tags=["items"])
api_router.include_router(categories.router, prefix="/categories", tags=["categories"])
api_router.include_router(suppliers.router, prefix="/suppliers", tags=["suppliers"])
api_router.include_router(transactions.router, prefix="/transactions", tags=["transactions"])

View File

@ -0,0 +1,90 @@
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 import crud
from app.api import deps
from app.core.config import settings
from app.core.security import create_access_token
from app.models.user import User
from app.schemas.token import Token
from app.schemas.user import User as UserSchema
from app.schemas.user import UserCreate
router = APIRouter()
@router.post("/login", response_model=Token)
def login_access_token(
db: Session = Depends(deps.get_db),
form_data: OAuth2PasswordRequestForm = Depends(),
) -> Any:
"""
OAuth2 compatible token login, get an access token for future requests.
"""
user = crud.user.authenticate(
db, username=form_data.username, password=form_data.password
)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
elif not crud.user.is_active(user):
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.username, expires_delta=access_token_expires
),
"token_type": "bearer",
}
@router.post("/register", response_model=UserSchema, status_code=status.HTTP_201_CREATED)
def register_user(
*,
db: Session = Depends(deps.get_db),
user_in: UserCreate,
) -> Any:
"""
Register a new user.
"""
# Check if user with the same email already exists
user = crud.user.get_by_email(db, email=user_in.email)
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="User with this email already exists",
)
# Check if user with the same username already exists
user = crud.user.get_by_username(db, username=user_in.username)
if user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="User with this username already exists",
)
# Create new user (non-superuser by default)
user_in.is_superuser = False
user = crud.user.create(db, obj_in=user_in)
return user
@router.get("/me", response_model=UserSchema)
def read_users_me(
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Get current user information.
"""
return current_user

View File

@ -0,0 +1,126 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app import crud
from app.api import deps
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(deps.get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Retrieve categories.
"""
categories = crud.category.get_multi(db, skip=skip, limit=limit)
return categories
@router.post("/", response_model=Category, status_code=status.HTTP_201_CREATED)
def create_category(
*,
db: Session = Depends(deps.get_db),
category_in: CategoryCreate,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Create new category.
"""
# Check if category with the same name already exists
category = crud.category.get_by_name(db, name=category_in.name)
if category:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Category with this name already exists",
)
category = crud.category.create(db, obj_in=category_in)
return category
@router.get("/{category_id}", response_model=Category)
def read_category(
*,
db: Session = Depends(deps.get_db),
category_id: int,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Get category by ID.
"""
category = crud.category.get(db, id=category_id)
if not category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found",
)
return category
@router.put("/{category_id}", response_model=Category)
def update_category(
*,
db: Session = Depends(deps.get_db),
category_id: int,
category_in: CategoryUpdate,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Update a category.
"""
category = crud.category.get(db, id=category_id)
if not category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found",
)
# Check if name is being changed and if it already exists
if category_in.name and category_in.name != category.name:
existing_category = crud.category.get_by_name(db, name=category_in.name)
if existing_category and existing_category.id != category_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Category with this name already exists",
)
category = crud.category.update(db, db_obj=category, obj_in=category_in)
return category
@router.delete("/{category_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
def delete_category(
*,
db: Session = Depends(deps.get_db),
category_id: int,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Delete a category.
"""
category = crud.category.get(db, id=category_id)
if not category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found",
)
# Check if any items are using this category
items = crud.item.get_by_category(db, category_id=category_id, skip=0, limit=1)
if items:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Cannot delete category with associated items",
)
crud.category.remove(db, id=category_id)
return None

View File

@ -0,0 +1,212 @@
from typing import Any, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from app import crud
from app.api import deps
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(deps.get_db),
skip: int = 0,
limit: int = 100,
category_id: Optional[int] = None,
supplier_id: Optional[int] = None,
low_stock: Optional[bool] = False,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Retrieve inventory items with optional filtering.
"""
if low_stock:
items = crud.item.get_low_stock(db, skip=skip, limit=limit)
elif category_id:
items = crud.item.get_by_category(db, category_id=category_id, skip=skip, limit=limit)
elif supplier_id:
items = crud.item.get_by_supplier(db, supplier_id=supplier_id, skip=skip, limit=limit)
else:
items = crud.item.get_multi(db, skip=skip, limit=limit)
return items
@router.post("/", response_model=Item, status_code=status.HTTP_201_CREATED)
def create_item(
*,
db: Session = Depends(deps.get_db),
item_in: ItemCreate,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Create new inventory item.
"""
# Check if item with the same SKU or barcode already exists
if item_in.sku:
item = crud.item.get_by_sku(db, sku=item_in.sku)
if item:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Item with this SKU already exists",
)
if item_in.barcode:
item = crud.item.get_by_barcode(db, barcode=item_in.barcode)
if item:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Item with this barcode already exists",
)
# Validate category if provided
if item_in.category_id and not crud.category.get(db, id=item_in.category_id):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="The category doesn't exist",
)
# Validate supplier if provided
if item_in.supplier_id and not crud.supplier.get(db, id=item_in.supplier_id):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="The supplier doesn't exist",
)
item = crud.item.create(db, obj_in=item_in)
return item
@router.get("/{item_id}", response_model=Item)
def read_item(
*,
db: Session = Depends(deps.get_db),
item_id: int,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Get item by ID.
"""
item = crud.item.get(db, id=item_id)
if not item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
return item
@router.put("/{item_id}", response_model=Item)
def update_item(
*,
db: Session = Depends(deps.get_db),
item_id: int,
item_in: ItemUpdate,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Update an item.
"""
item = crud.item.get(db, id=item_id)
if not item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
# Check if SKU is being changed and if it already exists
if item_in.sku and item_in.sku != item.sku:
existing_item = crud.item.get_by_sku(db, sku=item_in.sku)
if existing_item and existing_item.id != item_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Item with this SKU already exists",
)
# Check if barcode is being changed and if it already exists
if item_in.barcode and item_in.barcode != item.barcode:
existing_item = crud.item.get_by_barcode(db, barcode=item_in.barcode)
if existing_item and existing_item.id != item_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Item with this barcode already exists",
)
# Validate category if provided
if item_in.category_id and not crud.category.get(db, id=item_in.category_id):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="The category doesn't exist",
)
# Validate supplier if provided
if item_in.supplier_id and not crud.supplier.get(db, id=item_in.supplier_id):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="The supplier doesn't exist",
)
item = crud.item.update(db, db_obj=item, obj_in=item_in)
return item
@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
def delete_item(
*,
db: Session = Depends(deps.get_db),
item_id: int,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Delete an item.
"""
item = crud.item.get(db, id=item_id)
if not item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
# Check if the item has any transactions
transactions = crud.transaction.get_by_item(db, item_id=item_id, skip=0, limit=1)
if transactions:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Cannot delete item with transaction history",
)
crud.item.remove(db, id=item_id)
return None
@router.get("/search/", response_model=List[Item])
def search_items(
*,
db: Session = Depends(deps.get_db),
name: Optional[str] = Query(None, min_length=1),
sku: Optional[str] = Query(None, min_length=1),
barcode: Optional[str] = Query(None, min_length=1),
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Search for items by name, SKU, or barcode.
"""
if sku:
item = crud.item.get_by_sku(db, sku=sku)
return [item] if item else []
if barcode:
item = crud.item.get_by_barcode(db, barcode=barcode)
return [item] if item else []
if name:
items = db.query(crud.item.model).filter(
crud.item.model.name.ilike(f"%{name}%")
).all()
return items
# If no search parameters are provided, return an empty list
return []

View File

@ -0,0 +1,127 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app import crud
from app.api import deps
from app.models.user import User
from app.schemas.supplier import Supplier, SupplierCreate, SupplierUpdate
router = APIRouter()
@router.get("/", response_model=List[Supplier])
def read_suppliers(
db: Session = Depends(deps.get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Retrieve suppliers.
"""
suppliers = crud.supplier.get_multi(db, skip=skip, limit=limit)
return suppliers
@router.post("/", response_model=Supplier, status_code=status.HTTP_201_CREATED)
def create_supplier(
*,
db: Session = Depends(deps.get_db),
supplier_in: SupplierCreate,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Create new supplier.
"""
# Check if supplier with the same email already exists (if email is provided)
if supplier_in.email:
supplier = crud.supplier.get_by_email(db, email=supplier_in.email)
if supplier:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Supplier with this email already exists",
)
supplier = crud.supplier.create(db, obj_in=supplier_in)
return supplier
@router.get("/{supplier_id}", response_model=Supplier)
def read_supplier(
*,
db: Session = Depends(deps.get_db),
supplier_id: int,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Get supplier by ID.
"""
supplier = crud.supplier.get(db, id=supplier_id)
if not supplier:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Supplier not found",
)
return supplier
@router.put("/{supplier_id}", response_model=Supplier)
def update_supplier(
*,
db: Session = Depends(deps.get_db),
supplier_id: int,
supplier_in: SupplierUpdate,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Update a supplier.
"""
supplier = crud.supplier.get(db, id=supplier_id)
if not supplier:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Supplier not found",
)
# Check if email is being changed and if it already exists
if supplier_in.email and supplier_in.email != supplier.email:
existing_supplier = crud.supplier.get_by_email(db, email=supplier_in.email)
if existing_supplier and existing_supplier.id != supplier_id:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Supplier with this email already exists",
)
supplier = crud.supplier.update(db, db_obj=supplier, obj_in=supplier_in)
return supplier
@router.delete("/{supplier_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
def delete_supplier(
*,
db: Session = Depends(deps.get_db),
supplier_id: int,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Delete a supplier.
"""
supplier = crud.supplier.get(db, id=supplier_id)
if not supplier:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Supplier not found",
)
# Check if any items are using this supplier
items = crud.item.get_by_supplier(db, supplier_id=supplier_id, skip=0, limit=1)
if items:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Cannot delete supplier with associated items",
)
crud.supplier.remove(db, id=supplier_id)
return None

View File

@ -0,0 +1,176 @@
from datetime import datetime, timedelta
from typing import Any, List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.orm import Session
from app import crud
from app.api import deps
from app.models.transaction import TransactionType
from app.models.user import User
from app.schemas.transaction import Transaction, TransactionCreate, TransactionUpdate
router = APIRouter()
@router.get("/", response_model=List[Transaction])
def read_transactions(
db: Session = Depends(deps.get_db),
skip: int = 0,
limit: int = 100,
item_id: Optional[int] = None,
transaction_type: Optional[TransactionType] = None,
start_date: Optional[datetime] = None,
end_date: Optional[datetime] = None,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Retrieve transactions with optional filtering.
"""
if item_id:
transactions = crud.transaction.get_by_item(db, item_id=item_id, skip=skip, limit=limit)
elif transaction_type:
transactions = crud.transaction.get_by_type(db, transaction_type=transaction_type, skip=skip, limit=limit)
elif start_date and end_date:
transactions = crud.transaction.get_by_date_range(
db, start_date=start_date, end_date=end_date, skip=skip, limit=limit
)
else:
transactions = crud.transaction.get_multi(db, skip=skip, limit=limit)
return transactions
@router.post("/", response_model=Transaction, status_code=status.HTTP_201_CREATED)
def create_transaction(
*,
db: Session = Depends(deps.get_db),
transaction_in: TransactionCreate,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Create new transaction (stock in or stock out).
"""
# Check if item exists
item = crud.item.get(db, id=transaction_in.item_id)
if not item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
# For stock out transactions, check if there's enough stock
if transaction_in.transaction_type == TransactionType.STOCK_OUT and item.quantity < transaction_in.quantity:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Insufficient stock. Available: {item.quantity}, Requested: {transaction_in.quantity}",
)
# Create transaction with item quantity update
transaction = crud.transaction.create_with_item_update(
db, obj_in=transaction_in, user_id=current_user.id
)
return transaction
@router.get("/{transaction_id}", response_model=Transaction)
def read_transaction(
*,
db: Session = Depends(deps.get_db),
transaction_id: int,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Get transaction by ID.
"""
transaction = crud.transaction.get(db, id=transaction_id)
if not transaction:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Transaction not found",
)
return transaction
@router.put("/{transaction_id}", response_model=Transaction)
def update_transaction(
*,
db: Session = Depends(deps.get_db),
transaction_id: int,
transaction_in: TransactionUpdate,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Update a transaction's metadata (notes, reference) only.
Quantity and item changes are not allowed to maintain inventory integrity.
"""
transaction = crud.transaction.get(db, id=transaction_id)
if not transaction:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Transaction not found",
)
# Only allow updating certain fields (reference, notes)
# Do not allow changing quantity, unit price to maintain inventory integrity
update_data = transaction_in.dict(exclude_unset=True)
for field in ["quantity", "unit_price"]:
if field in update_data:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Cannot update {field} after transaction is created",
)
# Calculate total_price if unit_price is updated
if "unit_price" in update_data:
update_data["total_price"] = transaction.quantity * update_data["unit_price"]
transaction = crud.transaction.update(db, db_obj=transaction, obj_in=update_data)
return transaction
@router.get("/by-item/{item_id}", response_model=List[Transaction])
def read_item_transactions(
*,
db: Session = Depends(deps.get_db),
item_id: int,
skip: int = 0,
limit: int = 100,
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Get all transactions for a specific item.
"""
# Check if item exists
item = crud.item.get(db, id=item_id)
if not item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
transactions = crud.transaction.get_by_item(db, item_id=item_id, skip=skip, limit=limit)
return transactions
@router.get("/report/daily/", response_model=List[Transaction])
def get_daily_transactions(
*,
db: Session = Depends(deps.get_db),
date: Optional[datetime] = Query(None),
current_user: User = Depends(deps.get_current_user),
) -> Any:
"""
Get all transactions for a specific day.
"""
# Default to today if no date is provided
if not date:
date = datetime.now()
start_date = datetime(date.year, date.month, date.day, 0, 0, 0)
end_date = start_date + timedelta(days=1) - timedelta(microseconds=1)
transactions = crud.transaction.get_by_date_range(
db, start_date=start_date, end_date=end_date
)
return transactions

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

@ -0,0 +1,31 @@
import os
from pathlib import Path
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
"""
Application settings class that loads configuration from environment variables.
"""
# API details
API_V1_STR: str = "/api/v1"
PROJECT_NAME: str = "Small Business Inventory Management System"
PROJECT_DESCRIPTION: str = "API for managing inventory in small businesses"
PROJECT_VERSION: str = "0.1.0"
# Security
SECRET_KEY: str = os.getenv("SECRET_KEY", "super-secret-key-for-development-only")
ACCESS_TOKEN_EXPIRE_MINUTES: int = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "60"))
ALGORITHM: str = "HS256"
# Database
DB_DIR = Path("/app") / "storage" / "db"
DB_DIR.mkdir(parents=True, exist_ok=True)
SQLALCHEMY_DATABASE_URL: str = f"sqlite:///{DB_DIR}/db.sqlite"
class Config:
case_sensitive = True
settings = Settings()

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 its 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 @@
from app.crud.crud_category import category
from app.crud.crud_item import item
from app.crud.crud_supplier import supplier
from app.crud.crud_transaction import transaction
from app.crud.crud_user import user

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

@ -0,0 +1,79 @@
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.db.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]):
"""
Initialize with SQLAlchemy model class.
"""
self.model = model
def get(self, db: Session, id: Any) -> Optional[ModelType]:
"""
Get a 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 with pagination.
"""
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.dict(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

21
app/crud/crud_category.py Normal file
View File

@ -0,0 +1,21 @@
from typing import 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()
category = CRUDCategory(Category)

69
app/crud/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_barcode(self, db: Session, *, barcode: str) -> Optional[Item]:
"""
Get an item by barcode.
"""
return db.query(Item).filter(Item.barcode == barcode).first()
def get_by_category(
self, db: Session, *, category_id: int, skip: int = 0, limit: int = 100
) -> List[Item]:
"""
Get items by category.
"""
return (
db.query(Item)
.filter(Item.category_id == category_id)
.offset(skip)
.limit(limit)
.all()
)
def get_by_supplier(
self, db: Session, *, supplier_id: int, skip: int = 0, limit: int = 100
) -> List[Item]:
"""
Get items by supplier.
"""
return (
db.query(Item)
.filter(Item.supplier_id == supplier_id)
.offset(skip)
.limit(limit)
.all()
)
def get_low_stock(
self, db: Session, *, skip: int = 0, limit: int = 100
) -> List[Item]:
"""
Get items with stock quantity below reorder level.
"""
return (
db.query(Item)
.filter(Item.quantity <= Item.reorder_level)
.offset(skip)
.limit(limit)
.all()
)
item = CRUDItem(Item)

27
app/crud/crud_supplier.py Normal file
View File

@ -0,0 +1,27 @@
from typing import Optional
from sqlalchemy.orm import Session
from app.crud.base import CRUDBase
from app.models.supplier import Supplier
from app.schemas.supplier import SupplierCreate, SupplierUpdate
class CRUDSupplier(CRUDBase[Supplier, SupplierCreate, SupplierUpdate]):
"""
CRUD operations for Supplier model.
"""
def get_by_name(self, db: Session, *, name: str) -> Optional[Supplier]:
"""
Get a supplier by name.
"""
return db.query(Supplier).filter(Supplier.name == name).first()
def get_by_email(self, db: Session, *, email: str) -> Optional[Supplier]:
"""
Get a supplier by email.
"""
return db.query(Supplier).filter(Supplier.email == email).first()
supplier = CRUDSupplier(Supplier)

View File

@ -0,0 +1,125 @@
from datetime import datetime
from typing import List
from sqlalchemy import and_
from sqlalchemy.orm import Session
from app.crud.base import CRUDBase
from app.models.item import Item
from app.models.transaction import Transaction, TransactionType
from app.schemas.transaction import TransactionCreate, TransactionUpdate
class CRUDTransaction(CRUDBase[Transaction, TransactionCreate, TransactionUpdate]):
"""
CRUD operations for Transaction model.
"""
def create_with_item_update(
self, db: Session, *, obj_in: TransactionCreate, user_id: int
) -> Transaction:
"""
Create a transaction and update the associated item's quantity.
"""
# Get the item to update its quantity
item = db.query(Item).filter(Item.id == obj_in.item_id).first()
if not item:
raise ValueError(f"Item with ID {obj_in.item_id} not found")
# Calculate the total price
total_price = obj_in.quantity * obj_in.unit_price
# Create transaction object
db_obj = Transaction(
item_id=obj_in.item_id,
user_id=user_id,
quantity=obj_in.quantity,
transaction_type=obj_in.transaction_type,
unit_price=obj_in.unit_price,
total_price=total_price,
reference=obj_in.reference,
notes=obj_in.notes,
timestamp=datetime.utcnow()
)
# Update the item quantity based on transaction type
if obj_in.transaction_type == TransactionType.STOCK_IN:
item.quantity += obj_in.quantity
elif obj_in.transaction_type == TransactionType.STOCK_OUT:
if item.quantity < obj_in.quantity:
raise ValueError(f"Insufficient stock. Available: {item.quantity}, Requested: {obj_in.quantity}")
item.quantity -= obj_in.quantity
# Save to database
db.add(db_obj)
db.add(item)
db.commit()
db.refresh(db_obj)
return db_obj
def get_by_item(
self, db: Session, *, item_id: int, skip: int = 0, limit: int = 100
) -> List[Transaction]:
"""
Get transactions for a specific item.
"""
return (
db.query(Transaction)
.filter(Transaction.item_id == item_id)
.order_by(Transaction.timestamp.desc())
.offset(skip)
.limit(limit)
.all()
)
def get_by_user(
self, db: Session, *, user_id: int, skip: int = 0, limit: int = 100
) -> List[Transaction]:
"""
Get transactions performed by a specific user.
"""
return (
db.query(Transaction)
.filter(Transaction.user_id == user_id)
.order_by(Transaction.timestamp.desc())
.offset(skip)
.limit(limit)
.all()
)
def get_by_date_range(
self, db: Session, *, start_date: datetime, end_date: datetime,
skip: int = 0, limit: int = 100
) -> List[Transaction]:
"""
Get transactions within a date range.
"""
return (
db.query(Transaction)
.filter(and_(
Transaction.timestamp >= start_date,
Transaction.timestamp <= end_date
))
.order_by(Transaction.timestamp.desc())
.offset(skip)
.limit(limit)
.all()
)
def get_by_type(
self, db: Session, *, transaction_type: TransactionType,
skip: int = 0, limit: int = 100
) -> List[Transaction]:
"""
Get transactions by type (stock in or stock out).
"""
return (
db.query(Transaction)
.filter(Transaction.transaction_type == transaction_type)
.order_by(Transaction.timestamp.desc())
.offset(skip)
.limit(limit)
.all()
)
transaction = CRUDTransaction(Transaction)

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

@ -0,0 +1,84 @@
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 get_by_username(self, db: Session, *, username: str) -> Optional[User]:
"""
Get a user by username.
"""
return db.query(User).filter(User.username == username).first()
def create(self, db: Session, *, obj_in: UserCreate) -> User:
"""
Create a new user with hashed password.
"""
db_obj = User(
username=obj_in.username,
email=obj_in.email,
hashed_password=get_password_hash(obj_in.password),
full_name=obj_in.full_name,
is_active=obj_in.is_active,
is_superuser=obj_in.is_superuser,
)
db.add(db_obj)
db.commit()
db.refresh(db_obj)
return db_obj
def update(
self, db: Session, *, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]
) -> User:
"""
Update a user, handling password hashing if provided.
"""
if isinstance(obj_in, dict):
update_data = obj_in
else:
update_data = obj_in.dict(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, *, username: str, password: str) -> Optional[User]:
"""
Authenticate a user by username and password.
"""
user = self.get_by_username(db, username=username)
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
user = CRUDUser(User)

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

@ -0,0 +1,3 @@
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

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

@ -0,0 +1,52 @@
from typing import Generator
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.orm import Session
from app.core.config import settings
from app.db.session import SessionLocal
from app.models.user import User
from app.schemas.token import TokenPayload
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
def get_db() -> Generator:
"""
Get a database session as a dependency for routes.
"""
db = SessionLocal()
try:
yield db
finally:
db.close()
def get_current_user(
db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> User:
"""
Get the current user based on the JWT token.
"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
)
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenPayload(username=username)
except JWTError:
raise credentials_exception
user = db.query(User).filter(User.username == token_data.username).first()
if user is None:
raise credentials_exception
return user

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

@ -0,0 +1,11 @@
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
engine = create_engine(
settings.SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False} # For SQLite only
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

17
app/models/base_model.py Normal file
View File

@ -0,0 +1,17 @@
from app.db.base import Base
from app.models.category import Category
from app.models.item import Item
from app.models.supplier import Supplier
from app.models.transaction import Transaction, TransactionType
from app.models.user import User
# Import all models here to make them available for Alembic
__all__ = [
"Base",
"User",
"Item",
"Category",
"Supplier",
"Transaction",
"TransactionType"
]

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

@ -0,0 +1,18 @@
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship
from app.db.base import Base
class Category(Base):
"""
Category model for categorizing inventory items.
"""
__tablename__ = "categories"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True, nullable=False, unique=True)
description = Column(String, nullable=True)
# Relationships
items = relationship("Item", back_populates="category")

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

@ -0,0 +1,27 @@
from sqlalchemy import Column, Float, ForeignKey, Integer, String, Text
from sqlalchemy.orm import relationship
from app.db.base import Base
class Item(Base):
"""
Inventory item model for tracking products and stock.
"""
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True, nullable=False)
description = Column(Text, nullable=True)
sku = Column(String, unique=True, index=True, nullable=True)
barcode = Column(String, unique=True, index=True, nullable=True)
quantity = Column(Integer, default=0, nullable=False)
unit_price = Column(Float, nullable=False)
reorder_level = Column(Integer, default=10, nullable=False)
category_id = Column(Integer, ForeignKey("categories.id"), nullable=True)
supplier_id = Column(Integer, ForeignKey("suppliers.id"), nullable=True)
# Relationships
category = relationship("Category", back_populates="items")
supplier = relationship("Supplier", back_populates="items")
transactions = relationship("Transaction", back_populates="item")

21
app/models/supplier.py Normal file
View File

@ -0,0 +1,21 @@
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship
from app.db.base import Base
class Supplier(Base):
"""
Supplier model for tracking inventory suppliers.
"""
__tablename__ = "suppliers"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True, nullable=False)
contact_name = Column(String, nullable=True)
email = Column(String, nullable=True)
phone = Column(String, nullable=True)
address = Column(String, nullable=True)
# Relationships
items = relationship("Item", back_populates="supplier")

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

@ -0,0 +1,39 @@
from datetime import datetime
import enum
from sqlalchemy import Column, DateTime, Enum, Float, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from app.db.base import Base
class TransactionType(str, enum.Enum):
"""
Enum for transaction types (stock in or out).
"""
STOCK_IN = "stock_in"
STOCK_OUT = "stock_out"
class Transaction(Base):
"""
Transaction model for tracking inventory movements.
"""
__tablename__ = "transactions"
id = Column(Integer, primary_key=True, index=True)
quantity = Column(Integer, nullable=False)
transaction_type = Column(Enum(TransactionType), nullable=False)
unit_price = Column(Float, nullable=False)
total_price = Column(Float, nullable=False)
reference = Column(String, nullable=True)
notes = Column(String, nullable=True)
timestamp = Column(DateTime, default=datetime.utcnow, nullable=False)
# Foreign keys
item_id = Column(Integer, ForeignKey("items.id"), nullable=False)
user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
# Relationships
item = relationship("Item", back_populates="transactions")
user = relationship("User", back_populates="transactions")

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

@ -0,0 +1,22 @@
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import relationship
from app.db.base import Base
class User(Base):
"""
User model for authentication and authorization.
"""
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True, nullable=False)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
full_name = Column(String)
is_active = Column(Boolean, default=True)
is_superuser = Column(Boolean, default=False)
# Relationships
transactions = relationship("Transaction", back_populates="user")

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

@ -0,0 +1,51 @@
from typing import List, Optional
from pydantic import BaseModel, Field
class CategoryBase(BaseModel):
"""
Base category schema with common attributes.
"""
name: str = Field(..., description="Category name")
description: Optional[str] = Field(None, description="Category description")
class CategoryCreate(CategoryBase):
"""
Schema for category creation.
"""
pass
class CategoryUpdate(BaseModel):
"""
Schema for updating category information.
"""
name: Optional[str] = Field(None, description="Category name")
description: Optional[str] = Field(None, description="Category description")
class CategoryInDBBase(CategoryBase):
"""
Base schema for categories from the database.
"""
id: int = Field(..., description="Category ID")
class Config:
from_attributes = True
class Category(CategoryInDBBase):
"""
Schema for category information returned to clients.
"""
pass
class CategoryWithItems(CategoryInDBBase):
"""
Schema for category with related items.
"""
from app.schemas.item import Item
items: List[Item] = Field([], description="Items in this category")

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

@ -0,0 +1,68 @@
from typing import Optional
from pydantic import BaseModel, Field
class ItemBase(BaseModel):
"""
Base item schema with common attributes.
"""
name: str = Field(..., description="Item name")
description: Optional[str] = Field(None, description="Item description")
sku: Optional[str] = Field(None, description="Stock Keeping Unit")
barcode: Optional[str] = Field(None, description="Barcode/UPC")
quantity: int = Field(0, description="Current stock quantity", ge=0)
unit_price: float = Field(..., description="Unit price", gt=0)
reorder_level: int = Field(10, description="Quantity at which to reorder", ge=0)
category_id: Optional[int] = Field(None, description="ID of the category")
supplier_id: Optional[int] = Field(None, description="ID of the supplier")
class ItemCreate(ItemBase):
"""
Schema for item creation.
"""
pass
class ItemUpdate(BaseModel):
"""
Schema for updating item information.
"""
name: Optional[str] = Field(None, description="Item name")
description: Optional[str] = Field(None, description="Item description")
sku: Optional[str] = Field(None, description="Stock Keeping Unit")
barcode: Optional[str] = Field(None, description="Barcode/UPC")
quantity: Optional[int] = Field(None, description="Current stock quantity", ge=0)
unit_price: Optional[float] = Field(None, description="Unit price", gt=0)
reorder_level: Optional[int] = Field(None, description="Quantity at which to reorder", ge=0)
category_id: Optional[int] = Field(None, description="ID of the category")
supplier_id: Optional[int] = Field(None, description="ID of the supplier")
class ItemInDBBase(ItemBase):
"""
Base schema for items from the database.
"""
id: int = Field(..., description="Item ID")
class Config:
from_attributes = True
class Item(ItemInDBBase):
"""
Schema for item information returned to clients.
"""
pass
class ItemWithRelations(ItemInDBBase):
"""
Schema for item with related entities.
"""
from app.schemas.category import Category
from app.schemas.supplier import Supplier
category: Optional[Category] = Field(None, description="Item category")
supplier: Optional[Supplier] = Field(None, description="Item supplier")

57
app/schemas/supplier.py Normal file
View File

@ -0,0 +1,57 @@
from typing import List, Optional
from pydantic import BaseModel, EmailStr, Field
class SupplierBase(BaseModel):
"""
Base supplier schema with common attributes.
"""
name: str = Field(..., description="Supplier name")
contact_name: Optional[str] = Field(None, description="Contact person name")
email: Optional[EmailStr] = Field(None, description="Supplier email address")
phone: Optional[str] = Field(None, description="Supplier phone number")
address: Optional[str] = Field(None, description="Supplier address")
class SupplierCreate(SupplierBase):
"""
Schema for supplier creation.
"""
pass
class SupplierUpdate(BaseModel):
"""
Schema for updating supplier information.
"""
name: Optional[str] = Field(None, description="Supplier name")
contact_name: Optional[str] = Field(None, description="Contact person name")
email: Optional[EmailStr] = Field(None, description="Supplier email address")
phone: Optional[str] = Field(None, description="Supplier phone number")
address: Optional[str] = Field(None, description="Supplier address")
class SupplierInDBBase(SupplierBase):
"""
Base schema for suppliers from the database.
"""
id: int = Field(..., description="Supplier ID")
class Config:
from_attributes = True
class Supplier(SupplierInDBBase):
"""
Schema for supplier information returned to clients.
"""
pass
class SupplierWithItems(SupplierInDBBase):
"""
Schema for supplier with related items.
"""
from app.schemas.item import Item
items: List[Item] = Field([], description="Items from this supplier")

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

@ -0,0 +1,18 @@
from typing import Optional
from pydantic import BaseModel, Field
class Token(BaseModel):
"""
Schema for the token response.
"""
access_token: str = Field(..., description="JWT access token")
token_type: str = Field(..., description="Token type (bearer)")
class TokenPayload(BaseModel):
"""
Schema for the token payload (JWT claims).
"""
username: Optional[str] = None

View File

@ -0,0 +1,71 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field, validator
from app.models.transaction import TransactionType
class TransactionBase(BaseModel):
"""
Base transaction schema with common attributes.
"""
item_id: int = Field(..., description="ID of the item")
quantity: int = Field(..., description="Transaction quantity", gt=0)
transaction_type: TransactionType = Field(..., description="Type of transaction")
unit_price: float = Field(..., description="Unit price", gt=0)
reference: Optional[str] = Field(None, description="Reference number or order ID")
notes: Optional[str] = Field(None, description="Additional notes")
class TransactionCreate(TransactionBase):
"""
Schema for transaction creation.
"""
@validator('transaction_type')
def validate_transaction_type(cls, v):
if v not in [TransactionType.STOCK_IN, TransactionType.STOCK_OUT]:
raise ValueError('Transaction type must be either stock_in or stock_out')
return v
class TransactionUpdate(BaseModel):
"""
Schema for updating transaction information.
"""
quantity: Optional[int] = Field(None, description="Transaction quantity", gt=0)
unit_price: Optional[float] = Field(None, description="Unit price", gt=0)
reference: Optional[str] = Field(None, description="Reference number or order ID")
notes: Optional[str] = Field(None, description="Additional notes")
class TransactionInDBBase(TransactionBase):
"""
Base schema for transactions from the database.
"""
id: int = Field(..., description="Transaction ID")
total_price: float = Field(..., description="Total transaction price")
timestamp: datetime = Field(..., description="Transaction timestamp")
user_id: int = Field(..., description="ID of the user who performed the transaction")
class Config:
from_attributes = True
class Transaction(TransactionInDBBase):
"""
Schema for transaction information returned to clients.
"""
pass
class TransactionWithDetails(TransactionInDBBase):
"""
Schema for transaction with related entities.
"""
from app.schemas.item import Item
from app.schemas.user import User
item: Item = Field(..., description="Transaction item")
user: User = Field(..., description="User who performed the transaction")

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

@ -0,0 +1,57 @@
from typing import Optional
from pydantic import BaseModel, EmailStr, Field
class UserBase(BaseModel):
"""
Base user schema with common attributes.
"""
username: str = Field(..., description="Username for login")
email: EmailStr = Field(..., description="User's email address")
full_name: Optional[str] = Field(None, description="User's full name")
is_active: bool = Field(True, description="Whether the user is active")
is_superuser: bool = Field(False, description="Whether the user has admin privileges")
class UserCreate(UserBase):
"""
Schema for user creation with password.
"""
password: str = Field(..., description="User's password")
class UserUpdate(BaseModel):
"""
Schema for updating user information.
"""
username: Optional[str] = Field(None, description="Username for login")
email: Optional[EmailStr] = Field(None, description="User's email address")
full_name: Optional[str] = Field(None, description="User's full name")
password: Optional[str] = Field(None, description="User's password")
is_active: Optional[bool] = Field(None, description="Whether the user is active")
is_superuser: Optional[bool] = Field(None, description="Whether the user has admin privileges")
class UserInDBBase(UserBase):
"""
Base schema for users from the database.
"""
id: int = Field(..., description="User ID")
class Config:
from_attributes = True
class User(UserInDBBase):
"""
Schema for user information returned to clients.
"""
pass
class UserInDB(UserInDBBase):
"""
Schema for user in database including hashed password.
"""
hashed_password: str = Field(..., description="Hashed password")

53
main.py Normal file
View File

@ -0,0 +1,53 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
from app.api.v1.api import api_router
from app.core.config import settings
app = FastAPI(
title=settings.PROJECT_NAME,
description=settings.PROJECT_DESCRIPTION,
version=settings.PROJECT_VERSION,
openapi_url="/openapi.json",
docs_url="/docs",
redoc_url="/redoc",
)
# Add CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include API router
app.include_router(api_router, prefix=settings.API_V1_STR)
@app.get("/")
def root():
"""
Root endpoint that returns basic information about the service.
"""
return {
"title": settings.PROJECT_NAME,
"description": settings.PROJECT_DESCRIPTION,
"documentation": "/docs",
"health_check": "/health"
}
@app.get("/health", response_model=dict)
def health_check():
"""
Health check endpoint to verify the service is running.
"""
return {"status": "ok"}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

1
migrations/README Normal file
View File

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

78
migrations/env.py Normal file
View File

@ -0,0 +1,78 @@
from logging.config import fileConfig
from alembic import context
from sqlalchemy import engine_from_config, pool
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
# Interpret the config file for Python logging.
# This line sets up loggers basically.
fileConfig(config.config_file_name)
# add your model's MetaData object here
# for 'autogenerate' support
from app.models.base_model import Base
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
)
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,124 @@
"""Initial migration
Revision ID: 1a0c5d3a2b4c
Revises:
Create Date: 2023-09-21 00:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '1a0c5d3a2b4c'
down_revision = None
branch_labels = None
depends_on = None
def upgrade():
# Create users table
op.create_table(
'users',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('username', sa.String(), nullable=False),
sa.Column('email', sa.String(), nullable=False),
sa.Column('hashed_password', sa.String(), nullable=False),
sa.Column('full_name', sa.String(), nullable=True),
sa.Column('is_active', sa.Boolean(), nullable=True),
sa.Column('is_superuser', sa.Boolean(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False)
op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True)
# Create categories table
op.create_table(
'categories',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('description', sa.String(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_categories_id'), 'categories', ['id'], unique=False)
op.create_index(op.f('ix_categories_name'), 'categories', ['name'], unique=True)
# Create suppliers table
op.create_table(
'suppliers',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('contact_name', sa.String(), nullable=True),
sa.Column('email', sa.String(), nullable=True),
sa.Column('phone', sa.String(), nullable=True),
sa.Column('address', sa.String(), nullable=True),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_suppliers_id'), 'suppliers', ['id'], unique=False)
op.create_index(op.f('ix_suppliers_name'), 'suppliers', ['name'], unique=False)
# Create items table
op.create_table(
'items',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('sku', sa.String(), nullable=True),
sa.Column('barcode', sa.String(), nullable=True),
sa.Column('quantity', sa.Integer(), nullable=False),
sa.Column('unit_price', sa.Float(), nullable=False),
sa.Column('reorder_level', sa.Integer(), nullable=False),
sa.Column('category_id', sa.Integer(), nullable=True),
sa.Column('supplier_id', sa.Integer(), nullable=True),
sa.ForeignKeyConstraint(['category_id'], ['categories.id'], ),
sa.ForeignKeyConstraint(['supplier_id'], ['suppliers.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_items_barcode'), 'items', ['barcode'], unique=True)
op.create_index(op.f('ix_items_id'), 'items', ['id'], unique=False)
op.create_index(op.f('ix_items_name'), 'items', ['name'], unique=False)
op.create_index(op.f('ix_items_sku'), 'items', ['sku'], unique=True)
# Create transactions table
op.create_table(
'transactions',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('item_id', sa.Integer(), nullable=False),
sa.Column('user_id', sa.Integer(), nullable=False),
sa.Column('quantity', sa.Integer(), nullable=False),
sa.Column('transaction_type', sa.Enum('STOCK_IN', 'STOCK_OUT', name='transactiontype'), nullable=False),
sa.Column('unit_price', sa.Float(), nullable=False),
sa.Column('total_price', sa.Float(), nullable=False),
sa.Column('reference', sa.String(), nullable=True),
sa.Column('notes', sa.String(), nullable=True),
sa.Column('timestamp', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['item_id'], ['items.id'], ),
sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_transactions_id'), 'transactions', ['id'], unique=False)
def downgrade():
# Drop tables in reverse order
op.drop_index(op.f('ix_transactions_id'), table_name='transactions')
op.drop_table('transactions')
op.drop_index(op.f('ix_items_sku'), table_name='items')
op.drop_index(op.f('ix_items_name'), table_name='items')
op.drop_index(op.f('ix_items_id'), table_name='items')
op.drop_index(op.f('ix_items_barcode'), table_name='items')
op.drop_table('items')
op.drop_index(op.f('ix_suppliers_name'), table_name='suppliers')
op.drop_index(op.f('ix_suppliers_id'), table_name='suppliers')
op.drop_table('suppliers')
op.drop_index(op.f('ix_categories_name'), table_name='categories')
op.drop_index(op.f('ix_categories_id'), table_name='categories')
op.drop_table('categories')
op.drop_index(op.f('ix_users_username'), table_name='users')
op.drop_index(op.f('ix_users_id'), table_name='users')
op.drop_index(op.f('ix_users_email'), table_name='users')
op.drop_table('users')

30
pyproject.toml Normal file
View File

@ -0,0 +1,30 @@
[tool.ruff]
target-version = "py38"
line-length = 120
[tool.ruff.lint]
select = ["E", "F", "B", "I"]
ignore = ["B008", "E501", "B904", "E402"]
# Exclude a variety of commonly ignored directories.
exclude = [
".git",
".ruff_cache",
"__pycache__",
"venv",
".env",
".venv",
"env",
"dist",
]
# Allow unused variables when underscore-prefixed.
dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401"] # Ignore unused imports in __init__.py files
[tool.ruff.lint.isort]
force-single-line = false
known-first-party = ["app"]
force-sort-within-sections = true

23
requirements.txt Normal file
View File

@ -0,0 +1,23 @@
# Core dependencies
fastapi>=0.103.1
uvicorn>=0.23.2
pydantic>=2.3.0
pydantic-settings>=2.0.3
SQLAlchemy>=2.0.20
alembic>=1.12.0
# Database
aiosqlite>=0.19.0
# Authentication and security
python-jose>=3.3.0
passlib>=1.7.4
python-multipart>=0.0.6
bcrypt>=4.0.1
# Linting and code quality
ruff>=0.0.285
# Testing
pytest>=7.4.0
httpx>=0.24.1