Implement complete FastAPI inventory management system for small businesses

Features implemented:
- Product management with CRUD operations
- Category and supplier management
- Stock movement tracking with automatic updates
- Inventory reports and analytics
- SQLite database with Alembic migrations
- Health monitoring endpoints
- CORS configuration for API access
- Comprehensive API documentation
- Code quality with Ruff linting and formatting

The system provides a complete backend solution for small business inventory management with proper database relationships, stock tracking, and reporting capabilities.
This commit is contained in:
Automated Action 2025-07-23 09:07:46 +00:00
parent 2950aab4de
commit 447799c9a0
27 changed files with 1080 additions and 2 deletions

130
README.md
View File

@ -1,3 +1,129 @@
# FastAPI Application
# Small Business Inventory System
This is a FastAPI application bootstrapped by BackendIM, the AI-powered backend generation platform.
A comprehensive FastAPI-based inventory management system designed for small businesses to track products, manage stock levels, and generate reports.
## Features
- **Product Management**: Create, update, delete, and view products with detailed information
- **Category Management**: Organize products into categories
- **Supplier Management**: Track supplier information and relationships
- **Stock Management**: Add/remove stock with automatic movement tracking
- **Inventory Reports**: Get insights into stock levels, low stock alerts, and movement summaries
- **Health Monitoring**: Built-in health check endpoints
- **API Documentation**: Auto-generated OpenAPI/Swagger documentation
## Tech Stack
- **Backend**: FastAPI (Python)
- **Database**: SQLite with SQLAlchemy ORM
- **Migrations**: Alembic
- **Code Quality**: Ruff (linting and formatting)
- **Server**: Uvicorn
## Installation
1. Install dependencies:
```bash
pip install -r requirements.txt
```
2. Run database migrations:
```bash
alembic upgrade head
```
3. Start the server:
```bash
uvicorn main:app --host 0.0.0.0 --port 8000
```
## API Endpoints
### Root & Health
- `GET /` - API information and links
- `GET /health` - Health check endpoint
### Products
- `GET /api/v1/products/` - List all products (with filtering)
- `GET /api/v1/products/{id}` - Get product by ID
- `POST /api/v1/products/` - Create new product
- `PUT /api/v1/products/{id}` - Update product
- `DELETE /api/v1/products/{id}` - Delete product
- `POST /api/v1/products/{id}/stock/add` - Add stock
- `POST /api/v1/products/{id}/stock/remove` - Remove stock
### Categories
- `GET /api/v1/categories/` - List all categories
- `GET /api/v1/categories/{id}` - Get category by ID
- `POST /api/v1/categories/` - Create new category
- `PUT /api/v1/categories/{id}` - Update category
- `DELETE /api/v1/categories/{id}` - Delete category
### Suppliers
- `GET /api/v1/suppliers/` - List all suppliers
- `GET /api/v1/suppliers/{id}` - Get supplier by ID
- `POST /api/v1/suppliers/` - Create new supplier
- `PUT /api/v1/suppliers/{id}` - Update supplier
- `DELETE /api/v1/suppliers/{id}` - Delete supplier
### Stock Movements
- `GET /api/v1/stock-movements/` - List stock movements
- `GET /api/v1/stock-movements/{id}` - Get movement by ID
- `POST /api/v1/stock-movements/` - Create stock movement
### Reports
- `GET /api/v1/reports/inventory-summary` - Overall inventory statistics
- `GET /api/v1/reports/low-stock` - Products with low stock levels
- `GET /api/v1/reports/stock-movements-summary` - Stock movement statistics
## Documentation
- **API Documentation**: `/docs` (Swagger UI)
- **Alternative Docs**: `/redoc` (ReDoc)
- **OpenAPI JSON**: `/openapi.json`
## Database
The application uses SQLite database located at `/app/storage/db/db.sqlite`. The database schema includes:
- **Categories**: Product categories
- **Suppliers**: Supplier information
- **Products**: Product details with category and supplier relationships
- **Stock Movements**: Track all stock changes (IN/OUT/ADJUSTMENT)
## Environment Configuration
The application is configured to work with the following directory structure:
- Database: `/app/storage/db/`
- Storage: `/app/storage/` (for any file storage needs)
## Development
### Code Quality
```bash
# Lint and fix code
ruff check . --fix
# Format code
ruff format .
```
### Database Migrations
```bash
# Create new migration
alembic revision --autogenerate -m "migration message"
# Apply migrations
alembic upgrade head
```
## CORS Configuration
The API is configured with CORS to allow all origins (`*`) for development purposes. Adjust this in production as needed.
## Health Check
The `/health` endpoint provides information about:
- Application status
- Database connectivity
- Storage directory status

41
alembic.ini Normal file
View File

@ -0,0 +1,41 @@
[alembic]
script_location = alembic
prepend_sys_path = .
version_path_separator = os
sqlalchemy.url = sqlite:////app/storage/db/db.sqlite
[post_write_hooks]
[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

50
alembic/env.py Normal file
View File

@ -0,0 +1,50 @@
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
from app.db.base import Base
config = context.config
if config.config_file_name is not None:
fileConfig(config.config_file_name)
target_metadata = Base.metadata
def run_migrations_offline() -> None:
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() -> None:
connectable = engine_from_config(
config.get_section(config.config_ini_section),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
if context.is_offline_mode():
run_migrations_offline()
else:
run_migrations_online()

24
alembic/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() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}

View File

@ -0,0 +1,110 @@
"""Initial migration
Revision ID: 001
Revises:
Create Date: 2024-01-01 00:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
revision = "001"
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"categories",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("name", sa.String(length=100), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("created_at", sa.DateTime(), nullable=True),
sa.Column("updated_at", sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(op.f("ix_categories_id"), "categories", ["id"], unique=False)
op.create_index(op.f("ix_categories_name"), "categories", ["name"], unique=True)
op.create_table(
"suppliers",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("name", sa.String(length=200), nullable=False),
sa.Column("contact_person", sa.String(length=100), nullable=True),
sa.Column("email", sa.String(length=100), nullable=True),
sa.Column("phone", sa.String(length=20), nullable=True),
sa.Column("address", sa.Text(), nullable=True),
sa.Column("created_at", sa.DateTime(), nullable=True),
sa.Column("updated_at", sa.DateTime(), nullable=True),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(op.f("ix_suppliers_id"), "suppliers", ["id"], unique=False)
op.create_index(op.f("ix_suppliers_name"), "suppliers", ["name"], unique=False)
op.create_table(
"products",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("name", sa.String(length=200), nullable=False),
sa.Column("sku", sa.String(length=50), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("price", sa.Float(), nullable=False),
sa.Column("cost", sa.Float(), nullable=True),
sa.Column("quantity_in_stock", sa.Integer(), nullable=True),
sa.Column("minimum_stock_level", sa.Integer(), nullable=True),
sa.Column("category_id", sa.Integer(), nullable=True),
sa.Column("supplier_id", sa.Integer(), nullable=True),
sa.Column("created_at", sa.DateTime(), nullable=True),
sa.Column("updated_at", sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(
["category_id"],
["categories.id"],
),
sa.ForeignKeyConstraint(
["supplier_id"],
["suppliers.id"],
),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(op.f("ix_products_id"), "products", ["id"], unique=False)
op.create_index(op.f("ix_products_name"), "products", ["name"], unique=False)
op.create_index(op.f("ix_products_sku"), "products", ["sku"], unique=True)
op.create_table(
"stock_movements",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("product_id", sa.Integer(), nullable=False),
sa.Column(
"movement_type",
sa.Enum("IN", "OUT", "ADJUSTMENT", name="movementtype"),
nullable=False,
),
sa.Column("quantity", sa.Integer(), nullable=False),
sa.Column("reference", sa.String(length=100), nullable=True),
sa.Column("notes", sa.Text(), nullable=True),
sa.Column("created_at", sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(
["product_id"],
["products.id"],
),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(
op.f("ix_stock_movements_id"), "stock_movements", ["id"], unique=False
)
def downgrade() -> None:
op.drop_index(op.f("ix_stock_movements_id"), table_name="stock_movements")
op.drop_table("stock_movements")
op.drop_index(op.f("ix_products_sku"), table_name="products")
op.drop_index(op.f("ix_products_name"), table_name="products")
op.drop_index(op.f("ix_products_id"), table_name="products")
op.drop_table("products")
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")

0
app/__init__.py Normal file
View File

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

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

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

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

@ -0,0 +1,22 @@
from pathlib import Path
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
DB_DIR = Path("/app") / "storage" / "db"
DB_DIR.mkdir(parents=True, exist_ok=True)
SQLALCHEMY_DATABASE_URL = f"sqlite:///{DB_DIR}/db.sqlite"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

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

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

@ -0,0 +1,13 @@
from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime, Text
from app.db.base import Base
class Category(Base):
__tablename__ = "categories"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), unique=True, nullable=False, index=True)
description = Column(Text, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

24
app/models/product.py Normal file
View File

@ -0,0 +1,24 @@
from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime, Text, Float, ForeignKey
from sqlalchemy.orm import relationship
from app.db.base import Base
class Product(Base):
__tablename__ = "products"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(200), nullable=False, index=True)
sku = Column(String(50), unique=True, nullable=False, index=True)
description = Column(Text, nullable=True)
price = Column(Float, nullable=False)
cost = Column(Float, nullable=True)
quantity_in_stock = Column(Integer, default=0)
minimum_stock_level = Column(Integer, default=0)
category_id = Column(Integer, ForeignKey("categories.id"), nullable=True)
supplier_id = Column(Integer, ForeignKey("suppliers.id"), nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
category = relationship("Category", backref="products")
supplier = relationship("Supplier", backref="products")

View File

@ -0,0 +1,25 @@
from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime, Text, ForeignKey, Enum
from sqlalchemy.orm import relationship
from app.db.base import Base
import enum
class MovementType(enum.Enum):
IN = "in"
OUT = "out"
ADJUSTMENT = "adjustment"
class StockMovement(Base):
__tablename__ = "stock_movements"
id = Column(Integer, primary_key=True, index=True)
product_id = Column(Integer, ForeignKey("products.id"), nullable=False)
movement_type = Column(Enum(MovementType), nullable=False)
quantity = Column(Integer, nullable=False)
reference = Column(String(100), nullable=True)
notes = Column(Text, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
product = relationship("Product", backref="stock_movements")

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

@ -0,0 +1,16 @@
from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime, Text
from app.db.base import Base
class Supplier(Base):
__tablename__ = "suppliers"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(200), nullable=False, index=True)
contact_person = Column(String(100), nullable=True)
email = Column(String(100), nullable=True)
phone = Column(String(20), nullable=True)
address = Column(Text, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

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

81
app/routers/categories.py Normal file
View File

@ -0,0 +1,81 @@
from typing import List
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.models.category import Category
from app.schemas.category import (
Category as CategorySchema,
CategoryCreate,
CategoryUpdate,
)
router = APIRouter(prefix="/categories", tags=["categories"])
@router.get("/", response_model=List[CategorySchema])
def get_categories(
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
db: Session = Depends(get_db),
):
return db.query(Category).offset(skip).limit(limit).all()
@router.get("/{category_id}", response_model=CategorySchema)
def get_category(category_id: int, db: Session = Depends(get_db)):
category = db.query(Category).filter(Category.id == category_id).first()
if not category:
raise HTTPException(status_code=404, detail="Category not found")
return category
@router.post("/", response_model=CategorySchema)
def create_category(category: CategoryCreate, db: Session = Depends(get_db)):
existing_category = (
db.query(Category).filter(Category.name == category.name).first()
)
if existing_category:
raise HTTPException(status_code=400, detail="Category name already exists")
db_category = Category(**category.dict())
db.add(db_category)
db.commit()
db.refresh(db_category)
return db_category
@router.put("/{category_id}", response_model=CategorySchema)
def update_category(
category_id: int, category: CategoryUpdate, db: Session = Depends(get_db)
):
db_category = db.query(Category).filter(Category.id == category_id).first()
if not db_category:
raise HTTPException(status_code=404, detail="Category not found")
update_data = category.dict(exclude_unset=True)
if "name" in update_data:
existing_category = (
db.query(Category)
.filter(Category.name == update_data["name"], Category.id != category_id)
.first()
)
if existing_category:
raise HTTPException(status_code=400, detail="Category name already exists")
for field, value in update_data.items():
setattr(db_category, field, value)
db.commit()
db.refresh(db_category)
return db_category
@router.delete("/{category_id}")
def delete_category(category_id: int, db: Session = Depends(get_db)):
db_category = db.query(Category).filter(Category.id == category_id).first()
if not db_category:
raise HTTPException(status_code=404, detail="Category not found")
db.delete(db_category)
db.commit()
return {"message": "Category deleted successfully"}

153
app/routers/products.py Normal file
View File

@ -0,0 +1,153 @@
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.models.product import Product
from app.models.stock_movement import StockMovement, MovementType
from app.schemas.product import Product as ProductSchema, ProductCreate, ProductUpdate
router = APIRouter(prefix="/products", tags=["products"])
@router.get("/", response_model=List[ProductSchema])
def get_products(
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
category_id: Optional[int] = None,
supplier_id: Optional[int] = None,
low_stock: bool = False,
db: Session = Depends(get_db),
):
query = db.query(Product)
if category_id:
query = query.filter(Product.category_id == category_id)
if supplier_id:
query = query.filter(Product.supplier_id == supplier_id)
if low_stock:
query = query.filter(Product.quantity_in_stock <= Product.minimum_stock_level)
return query.offset(skip).limit(limit).all()
@router.get("/{product_id}", response_model=ProductSchema)
def get_product(product_id: int, db: Session = Depends(get_db)):
product = db.query(Product).filter(Product.id == product_id).first()
if not product:
raise HTTPException(status_code=404, detail="Product not found")
return product
@router.post("/", response_model=ProductSchema)
def create_product(product: ProductCreate, db: Session = Depends(get_db)):
existing_product = db.query(Product).filter(Product.sku == product.sku).first()
if existing_product:
raise HTTPException(status_code=400, detail="SKU already exists")
db_product = Product(**product.dict())
db.add(db_product)
db.commit()
db.refresh(db_product)
return db_product
@router.put("/{product_id}", response_model=ProductSchema)
def update_product(
product_id: int, product: ProductUpdate, db: Session = Depends(get_db)
):
db_product = db.query(Product).filter(Product.id == product_id).first()
if not db_product:
raise HTTPException(status_code=404, detail="Product not found")
update_data = product.dict(exclude_unset=True)
if "sku" in update_data:
existing_product = (
db.query(Product)
.filter(Product.sku == update_data["sku"], Product.id != product_id)
.first()
)
if existing_product:
raise HTTPException(status_code=400, detail="SKU already exists")
for field, value in update_data.items():
setattr(db_product, field, value)
db.commit()
db.refresh(db_product)
return db_product
@router.delete("/{product_id}")
def delete_product(product_id: int, db: Session = Depends(get_db)):
db_product = db.query(Product).filter(Product.id == product_id).first()
if not db_product:
raise HTTPException(status_code=404, detail="Product not found")
db.delete(db_product)
db.commit()
return {"message": "Product deleted successfully"}
@router.post("/{product_id}/stock/add")
def add_stock(
product_id: int,
quantity: int,
reference: Optional[str] = None,
notes: Optional[str] = None,
db: Session = Depends(get_db),
):
db_product = db.query(Product).filter(Product.id == product_id).first()
if not db_product:
raise HTTPException(status_code=404, detail="Product not found")
db_product.quantity_in_stock += quantity
movement = StockMovement(
product_id=product_id,
movement_type=MovementType.IN,
quantity=quantity,
reference=reference,
notes=notes,
)
db.add(movement)
db.commit()
db.refresh(db_product)
return {
"message": f"Added {quantity} units to stock",
"new_stock_level": db_product.quantity_in_stock,
}
@router.post("/{product_id}/stock/remove")
def remove_stock(
product_id: int,
quantity: int,
reference: Optional[str] = None,
notes: Optional[str] = None,
db: Session = Depends(get_db),
):
db_product = db.query(Product).filter(Product.id == product_id).first()
if not db_product:
raise HTTPException(status_code=404, detail="Product not found")
if db_product.quantity_in_stock < quantity:
raise HTTPException(status_code=400, detail="Insufficient stock")
db_product.quantity_in_stock -= quantity
movement = StockMovement(
product_id=product_id,
movement_type=MovementType.OUT,
quantity=quantity,
reference=reference,
notes=notes,
)
db.add(movement)
db.commit()
db.refresh(db_product)
return {
"message": f"Removed {quantity} units from stock",
"new_stock_level": db_product.quantity_in_stock,
}

79
app/routers/reports.py Normal file
View File

@ -0,0 +1,79 @@
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from sqlalchemy import func
from app.db.session import get_db
from app.models.product import Product
from app.models.stock_movement import StockMovement, MovementType
router = APIRouter(prefix="/reports", tags=["reports"])
@router.get("/inventory-summary")
def get_inventory_summary(db: Session = Depends(get_db)):
total_products = db.query(Product).count()
total_stock_value = (
db.query(func.sum(Product.price * Product.quantity_in_stock)).scalar() or 0
)
low_stock_products = (
db.query(Product)
.filter(Product.quantity_in_stock <= Product.minimum_stock_level)
.count()
)
return {
"total_products": total_products,
"total_stock_value": round(total_stock_value, 2),
"low_stock_products": low_stock_products,
}
@router.get("/low-stock")
def get_low_stock_products(db: Session = Depends(get_db)):
products = (
db.query(Product)
.filter(Product.quantity_in_stock <= Product.minimum_stock_level)
.all()
)
return [
{
"id": product.id,
"name": product.name,
"sku": product.sku,
"current_stock": product.quantity_in_stock,
"minimum_stock": product.minimum_stock_level,
"shortage": product.minimum_stock_level - product.quantity_in_stock,
}
for product in products
]
@router.get("/stock-movements-summary")
def get_stock_movements_summary(db: Session = Depends(get_db)):
stock_in = (
db.query(func.sum(StockMovement.quantity))
.filter(StockMovement.movement_type == MovementType.IN)
.scalar()
or 0
)
stock_out = (
db.query(func.sum(StockMovement.quantity))
.filter(StockMovement.movement_type == MovementType.OUT)
.scalar()
or 0
)
adjustments = (
db.query(func.sum(StockMovement.quantity))
.filter(StockMovement.movement_type == MovementType.ADJUSTMENT)
.scalar()
or 0
)
return {
"total_stock_in": stock_in,
"total_stock_out": stock_out,
"total_adjustments": adjustments,
"net_movement": stock_in - stock_out + adjustments,
}

View File

@ -0,0 +1,45 @@
from typing import List, Optional
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.models.stock_movement import StockMovement
from app.schemas.stock_movement import (
StockMovement as StockMovementSchema,
StockMovementCreate,
)
router = APIRouter(prefix="/stock-movements", tags=["stock-movements"])
@router.get("/", response_model=List[StockMovementSchema])
def get_stock_movements(
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
product_id: Optional[int] = None,
db: Session = Depends(get_db),
):
query = db.query(StockMovement)
if product_id:
query = query.filter(StockMovement.product_id == product_id)
return (
query.order_by(StockMovement.created_at.desc()).offset(skip).limit(limit).all()
)
@router.get("/{movement_id}", response_model=StockMovementSchema)
def get_stock_movement(movement_id: int, db: Session = Depends(get_db)):
movement = db.query(StockMovement).filter(StockMovement.id == movement_id).first()
if not movement:
raise HTTPException(status_code=404, detail="Stock movement not found")
return movement
@router.post("/", response_model=StockMovementSchema)
def create_stock_movement(movement: StockMovementCreate, db: Session = Depends(get_db)):
db_movement = StockMovement(**movement.dict())
db.add(db_movement)
db.commit()
db.refresh(db_movement)
return db_movement

66
app/routers/suppliers.py Normal file
View File

@ -0,0 +1,66 @@
from typing import List
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.models.supplier import Supplier
from app.schemas.supplier import (
Supplier as SupplierSchema,
SupplierCreate,
SupplierUpdate,
)
router = APIRouter(prefix="/suppliers", tags=["suppliers"])
@router.get("/", response_model=List[SupplierSchema])
def get_suppliers(
skip: int = Query(0, ge=0),
limit: int = Query(100, ge=1, le=1000),
db: Session = Depends(get_db),
):
return db.query(Supplier).offset(skip).limit(limit).all()
@router.get("/{supplier_id}", response_model=SupplierSchema)
def get_supplier(supplier_id: int, db: Session = Depends(get_db)):
supplier = db.query(Supplier).filter(Supplier.id == supplier_id).first()
if not supplier:
raise HTTPException(status_code=404, detail="Supplier not found")
return supplier
@router.post("/", response_model=SupplierSchema)
def create_supplier(supplier: SupplierCreate, db: Session = Depends(get_db)):
db_supplier = Supplier(**supplier.dict())
db.add(db_supplier)
db.commit()
db.refresh(db_supplier)
return db_supplier
@router.put("/{supplier_id}", response_model=SupplierSchema)
def update_supplier(
supplier_id: int, supplier: SupplierUpdate, db: Session = Depends(get_db)
):
db_supplier = db.query(Supplier).filter(Supplier.id == supplier_id).first()
if not db_supplier:
raise HTTPException(status_code=404, detail="Supplier not found")
update_data = supplier.dict(exclude_unset=True)
for field, value in update_data.items():
setattr(db_supplier, field, value)
db.commit()
db.refresh(db_supplier)
return db_supplier
@router.delete("/{supplier_id}")
def delete_supplier(supplier_id: int, db: Session = Depends(get_db)):
db_supplier = db.query(Supplier).filter(Supplier.id == supplier_id).first()
if not db_supplier:
raise HTTPException(status_code=404, detail="Supplier not found")
db.delete(db_supplier)
db.commit()
return {"message": "Supplier deleted successfully"}

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

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

@ -0,0 +1,26 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel
class CategoryBase(BaseModel):
name: str
description: Optional[str] = None
class CategoryCreate(CategoryBase):
pass
class CategoryUpdate(BaseModel):
name: Optional[str] = None
description: Optional[str] = None
class Category(CategoryBase):
id: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True

44
app/schemas/product.py Normal file
View File

@ -0,0 +1,44 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel
from app.schemas.category import Category
from app.schemas.supplier import Supplier
class ProductBase(BaseModel):
name: str
sku: str
description: Optional[str] = None
price: float
cost: Optional[float] = None
quantity_in_stock: int = 0
minimum_stock_level: int = 0
category_id: Optional[int] = None
supplier_id: Optional[int] = None
class ProductCreate(ProductBase):
pass
class ProductUpdate(BaseModel):
name: Optional[str] = None
sku: Optional[str] = None
description: Optional[str] = None
price: Optional[float] = None
cost: Optional[float] = None
quantity_in_stock: Optional[int] = None
minimum_stock_level: Optional[int] = None
category_id: Optional[int] = None
supplier_id: Optional[int] = None
class Product(ProductBase):
id: int
created_at: datetime
updated_at: datetime
category: Optional[Category] = None
supplier: Optional[Supplier] = None
class Config:
from_attributes = True

View File

@ -0,0 +1,26 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel
from app.models.stock_movement import MovementType
from app.schemas.product import Product
class StockMovementBase(BaseModel):
product_id: int
movement_type: MovementType
quantity: int
reference: Optional[str] = None
notes: Optional[str] = None
class StockMovementCreate(StockMovementBase):
pass
class StockMovement(StockMovementBase):
id: int
created_at: datetime
product: Optional[Product] = None
class Config:
from_attributes = True

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

@ -0,0 +1,32 @@
from datetime import datetime
from typing import Optional
from pydantic import BaseModel
class SupplierBase(BaseModel):
name: str
contact_person: Optional[str] = None
email: Optional[str] = None
phone: Optional[str] = None
address: Optional[str] = None
class SupplierCreate(SupplierBase):
pass
class SupplierUpdate(BaseModel):
name: Optional[str] = None
contact_person: Optional[str] = None
email: Optional[str] = None
phone: Optional[str] = None
address: Optional[str] = None
class Supplier(SupplierBase):
id: int
created_at: datetime
updated_at: datetime
class Config:
from_attributes = True

65
main.py Normal file
View File

@ -0,0 +1,65 @@
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pathlib import Path
from app.db.session import engine
from app.db.base import Base
from app.routers import products, categories, suppliers, stock_movements, reports
Base.metadata.create_all(bind=engine)
app = FastAPI(
title="Small Business Inventory System",
description="A comprehensive inventory management system for small businesses",
version="1.0.0",
openapi_url="/openapi.json",
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(products.router, prefix="/api/v1")
app.include_router(categories.router, prefix="/api/v1")
app.include_router(suppliers.router, prefix="/api/v1")
app.include_router(stock_movements.router, prefix="/api/v1")
app.include_router(reports.router, prefix="/api/v1")
@app.get("/")
def root():
return {
"title": "Small Business Inventory System",
"description": "A comprehensive inventory management system for small businesses",
"version": "1.0.0",
"documentation": "/docs",
"health_check": "/health",
}
@app.get("/health")
def health_check():
try:
db_dir = Path("/app/storage/db")
db_file = db_dir / "db.sqlite"
return {
"status": "healthy",
"database": {
"status": "connected",
"path": str(db_file),
"exists": db_file.exists(),
},
"storage": {"directory": str(db_dir), "exists": db_dir.exists()},
}
except Exception as e:
return {"status": "unhealthy", "error": str(e)}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

7
requirements.txt Normal file
View File

@ -0,0 +1,7 @@
fastapi==0.104.1
uvicorn[standard]==0.24.0
sqlalchemy==2.0.23
alembic==1.12.1
pydantic==2.5.0
python-multipart==0.0.6
ruff==0.1.6