backend.im/smallbusinessinventorymanagementsystem-jhmraf
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement Small Business Inventory Management System with FastAPI and SQLite

Browse Source
This commit is contained in:
Automated Action 2025-05-19 21:42:12 +00:00
parent 3b220168c8
commit 1b3ade7a03
46 changed files with 1904 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
README.md
View File

@ -1,3 +1,95 @@
# FastAPI Application
# Small Business Inventory Management System
This is a FastAPI application bootstrapped by BackendIM, the AI-powered backend generation platform.
A RESTful API for managing inventory for small businesses, built with FastAPI and SQLite.
## Features
- **Inventory Management**: Track items, categories, and suppliers
- **User Authentication**: Secure API endpoints with JWT authentication
- **Reporting**: Generate inventory reports
- **RESTful API**: Well-documented API with Swagger UI
## Technologies Used
- **FastAPI**: High-performance web framework
- **SQLAlchemy**: SQL toolkit and ORM
- **Alembic**: Database migration tool
- **SQLite**: Lightweight database
- **JWT**: Authentication mechanism
- **Pydantic**: Data validation
- **Uvicorn**: ASGI server
## API Endpoints
- **Authentication**: `/api/v1/auth/*`
- **Users**: `/api/v1/users/*`
- **Items**: `/api/v1/items/*`
- **Categories**: `/api/v1/categories/*`
- **Suppliers**: `/api/v1/suppliers/*`
- **Reports**: `/api/v1/reports/*`
- **Health Check**: `/health`
## Getting Started
### Prerequisites
- Python 3.7+
### Installation
1. Clone the repository
2. Install dependencies:
```bash
pip install -r requirements.txt
```
3. Run the migrations:
```bash
alembic upgrade head
```
4. Start the server:
```bash
uvicorn main:app --reload
```
5. Visit the API documentation at [http://localhost:8000/docs](http://localhost:8000/docs)
## Project Structure
```
├── app
│ ├── api
│ │ ├── api_v1
│ │ │ ├── endpoints
│ │ │ │ ├── auth.py
│ │ │ │ ├── categories.py
│ │ │ │ ├── items.py
│ │ │ │ ├── reports.py
│ │ │ │ ├── suppliers.py
│ │ │ │ └── users.py
│ │ │ └── api.py
│ │ └── deps.py
│ ├── core
│ │ ├── config.py
│ │ └── security.py
│ ├── crud
│ ├── db
│ │ └── session.py
│ ├── models
│ ├── schemas
│ └── utils
├── migrations
│ ├── versions
│ ├── env.py
│ └── script.py.mako
├── storage
│ └── db
├── alembic.ini
├── main.py
├── README.md
└── requirements.txt
```

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 example
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

0
app/__init__.py Normal file
View File

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

0
app/api/api_v1/__init__.py Normal file
View File

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

@ -0,0 +1,11 @@
from fastapi import APIRouter
from app.api.api_v1.endpoints import items, categories, suppliers, auth, users, reports
api_router = APIRouter()
api_router.include_router(auth.router, prefix="/auth", tags=["Authentication"])
api_router.include_router(users.router, prefix="/users", tags=["Users"])
api_router.include_router(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(reports.router, prefix="/reports", tags=["Reports"])

0
app/api/api_v1/endpoints/__init__.py Normal file
View File

43
app/api/api_v1/endpoints/auth.py Normal file
View File

@ -0,0 +1,43 @@
from datetime import timedelta
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from app.api.deps import get_db
from app.core.security import create_access_token
from app.core.config import settings
from app.crud import user
from app.schemas.token import Token
router = APIRouter()
@router.post("/login", response_model=Token)
def login_access_token(
db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()
) -> Any:
"""
OAuth2 compatible token login, get an access token for future requests.
"""
authenticated_user = user.authenticate(
db, username=form_data.username, password=form_data.password
)
if not authenticated_user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
)
if not user.is_active(authenticated_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(
authenticated_user.id, expires_delta=access_token_expires
),
"token_type": "bearer",
}

102
app/api/api_v1/endpoints/categories.py Normal file
View File

@ -0,0 +1,102 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.api.deps import get_db, get_current_user
from app.models.user import User
from app.crud import category
from app.schemas.category import Category, CategoryCreate, CategoryUpdate
router = APIRouter()
@router.get("/", response_model=List[Category])
def read_categories(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Retrieve categories.
"""
categories = category.get_multi(db, skip=skip, limit=limit)
return categories
@router.post("/", response_model=Category)
def create_category(
*,
category_in: CategoryCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Create new category.
"""
db_category = category.get_by_name(db, name=category_in.name)
if db_category:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="A category with this name already exists",
)
return category.create(db, obj_in=category_in)
@router.get("/{category_id}", response_model=Category)
def read_category(
*,
category_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get category by ID.
"""
db_category = category.get(db, id=category_id)
if not db_category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found",
)
return db_category
@router.put("/{category_id}", response_model=Category)
def update_category(
*,
category_id: int,
category_in: CategoryUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Update a category.
"""
db_category = category.get(db, id=category_id)
if not db_category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found",
)
return category.update(db, db_obj=db_category, obj_in=category_in)
@router.delete("/{category_id}", response_model=Category)
def delete_category(
*,
category_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Delete a category.
"""
db_category = category.get(db, id=category_id)
if not db_category:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Category not found",
)
return category.remove(db, id=category_id)

200
app/api/api_v1/endpoints/items.py Normal file
View File

@ -0,0 +1,200 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.api.deps import get_db, get_current_user
from app.models.user import User
from app.crud import item, transaction
from app.schemas.item import (
Item, ItemCreate, ItemUpdate,
Transaction, TransactionCreate,
InventoryAdjustment
)
router = APIRouter()
@router.get("/", response_model=List[Item])
def read_items(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
category_id: int = None,
supplier_id: int = None,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Retrieve items with optional filtering by category or supplier.
"""
if category_id:
items = item.get_by_category(db, category_id=category_id, skip=skip, limit=limit)
elif supplier_id:
items = item.get_by_supplier(db, supplier_id=supplier_id, skip=skip, limit=limit)
else:
items = item.get_multi(db, skip=skip, limit=limit)
return items
@router.get("/low-stock", response_model=List[Item])
def read_low_stock_items(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Retrieve items with stock level at or below reorder level.
"""
items = item.get_low_stock_items(db, skip=skip, limit=limit)
return items
@router.post("/", response_model=Item)
def create_item(
*,
item_in: ItemCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Create new item.
"""
db_item = item.get_by_sku(db, sku=item_in.sku)
if db_item:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="An item with this SKU already exists",
)
return item.create(db, obj_in=item_in)
@router.get("/{item_id}", response_model=Item)
def read_item(
*,
item_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get item by ID.
"""
db_item = item.get(db, id=item_id)
if not db_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
return db_item
@router.put("/{item_id}", response_model=Item)
def update_item(
*,
item_id: int,
item_in: ItemUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Update an item.
"""
db_item = item.get(db, id=item_id)
if not db_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
return item.update(db, db_obj=db_item, obj_in=item_in)
@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT, response_model=None)
def delete_item(
*,
item_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> None:
"""
Delete an item.
"""
db_item = item.get(db, id=item_id)
if not db_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
item.remove(db, id=item_id)
return None
@router.post("/{item_id}/adjust", response_model=Item)
def adjust_item_inventory(
*,
item_id: int,
adjustment: InventoryAdjustment,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Adjust the inventory quantity of an item.
"""
db_item = item.get(db, id=item_id)
if not db_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
# Ensure the adjustment doesn't result in negative stock
if db_item.quantity + adjustment.quantity < 0:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Adjustment would result in negative stock",
)
return item.adjust_inventory(db, item_id=item_id, adjustment=adjustment)
@router.get("/{item_id}/transactions", response_model=List[Transaction])
def read_item_transactions(
*,
item_id: int,
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get all transactions for a specific item.
"""
db_item = item.get(db, id=item_id)
if not db_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Item not found",
)
return transaction.get_transactions_by_item(db, item_id=item_id, skip=skip, limit=limit)
@router.post("/transactions", response_model=Transaction)
def create_transaction(
*,
transaction_in: TransactionCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Create a new transaction (purchase, sale, etc.).
"""
# Validate that all item IDs exist
for item_data in transaction_in.items:
db_item = item.get(db, id=item_data.item_id)
if not db_item:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Item with ID {item_data.item_id} not found",
)
return transaction.create(db, obj_in=transaction_in)

99
app/api/api_v1/endpoints/reports.py Normal file
View File

@ -0,0 +1,99 @@
from typing import Any
from fastapi import APIRouter, Depends
from sqlalchemy import func
from sqlalchemy.orm import Session
from app.api.deps import get_db, get_current_user
from app.models.user import User
from app.models.item import Item, Transaction
router = APIRouter()
@router.get("/inventory-summary")
def inventory_summary(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get inventory summary including total items, total value, and out-of-stock count.
"""
# Total number of items
total_items = db.query(func.count(Item.id)).scalar()
# Total inventory value
total_value = db.query(func.sum(Item.quantity * Item.unit_price)).scalar() or 0
# Out of stock items
out_of_stock_count = db.query(func.count(Item.id)).filter(Item.quantity == 0).scalar()
# Low stock items (below reorder level)
low_stock_count = db.query(func.count(Item.id)).filter(
Item.quantity > 0, Item.quantity <= Item.reorder_level
).scalar()
return {
"total_items": total_items,
"total_value": total_value,
"out_of_stock_count": out_of_stock_count,
"low_stock_count": low_stock_count,
}
@router.get("/category-summary")
def category_summary(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get inventory summary by category.
"""
query = db.query(
Item.category_id,
func.count(Item.id).label("item_count"),
func.sum(Item.quantity).label("total_quantity"),
func.sum(Item.quantity * Item.unit_price).label("total_value")
).group_by(Item.category_id).all()
return [
{
"category_id": category_id,
"item_count": item_count,
"total_quantity": total_quantity or 0,
"total_value": total_value or 0
}
for category_id, item_count, total_quantity, total_value in query
]
@router.get("/transaction-summary")
def transaction_summary(
db: Session = Depends(get_db),
start_date: str = None,
end_date: str = None,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get transaction summary by type, optionally filtered by date range.
"""
query = db.query(
Transaction.transaction_type,
func.count(Transaction.id).label("transaction_count")
).group_by(Transaction.transaction_type)
# Apply date filters if provided
if start_date:
query = query.filter(Transaction.created_at >= start_date)
if end_date:
query = query.filter(Transaction.created_at <= end_date)
result = query.all()
return [
{
"transaction_type": transaction_type,
"transaction_count": transaction_count
}
for transaction_type, transaction_count in result
]

102
app/api/api_v1/endpoints/suppliers.py Normal file
View File

@ -0,0 +1,102 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.api.deps import get_db, get_current_user
from app.models.user import User
from app.crud import supplier
from app.schemas.supplier import Supplier, SupplierCreate, SupplierUpdate
router = APIRouter()
@router.get("/", response_model=List[Supplier])
def read_suppliers(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Retrieve suppliers.
"""
suppliers = supplier.get_multi(db, skip=skip, limit=limit)
return suppliers
@router.post("/", response_model=Supplier)
def create_supplier(
*,
supplier_in: SupplierCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Create new supplier.
"""
db_supplier = supplier.get_by_name(db, name=supplier_in.name)
if db_supplier:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="A supplier with this name already exists",
)
return supplier.create(db, obj_in=supplier_in)
@router.get("/{supplier_id}", response_model=Supplier)
def read_supplier(
*,
supplier_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get supplier by ID.
"""
db_supplier = supplier.get(db, id=supplier_id)
if not db_supplier:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Supplier not found",
)
return db_supplier
@router.put("/{supplier_id}", response_model=Supplier)
def update_supplier(
*,
supplier_id: int,
supplier_in: SupplierUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Update a supplier.
"""
db_supplier = supplier.get(db, id=supplier_id)
if not db_supplier:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Supplier not found",
)
return supplier.update(db, db_obj=db_supplier, obj_in=supplier_in)
@router.delete("/{supplier_id}", response_model=Supplier)
def delete_supplier(
*,
supplier_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Delete a supplier.
"""
db_supplier = supplier.get(db, id=supplier_id)
if not db_supplier:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Supplier not found",
)
return supplier.remove(db, id=supplier_id)

134
app/api/api_v1/endpoints/users.py Normal file
View File

@ -0,0 +1,134 @@
from typing import Any, List
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from app.api.deps import get_db, get_current_user
from app.models.user import User
from app.crud import user
from app.schemas.user import User as UserSchema, UserCreate, UserUpdate
router = APIRouter()
@router.get("/me", response_model=UserSchema)
def read_current_user(
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get current user.
"""
return current_user
@router.put("/me", response_model=UserSchema)
def update_current_user(
*,
user_in: UserUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Update own user.
"""
updated_user = user.update(db, db_obj=current_user, obj_in=user_in)
return updated_user
@router.get("/", response_model=List[UserSchema])
def read_users(
db: Session = Depends(get_db),
skip: int = 0,
limit: int = 100,
current_user: User = Depends(get_current_user),
) -> Any:
"""
Retrieve users.
"""
if not user.is_superuser(current_user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
users = user.get_multi(db, skip=skip, limit=limit)
return users
@router.post("/", response_model=UserSchema)
def create_user(
*,
user_in: UserCreate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Create new user.
"""
if not user.is_superuser(current_user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
db_user = user.get_by_email(db, email=user_in.email)
if db_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="A user with this email already exists",
)
db_user = user.get_by_username(db, username=user_in.username)
if db_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="A user with this username already exists",
)
created_user = user.create(db, obj_in=user_in)
return created_user
@router.get("/{user_id}", response_model=UserSchema)
def read_user_by_id(
user_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Get a specific user by id.
"""
db_user = user.get(db, id=user_id)
if not db_user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
if db_user.id != current_user.id and not user.is_superuser(current_user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
return db_user
@router.put("/{user_id}", response_model=UserSchema)
def update_user(
*,
user_id: int,
user_in: UserUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> Any:
"""
Update a user.
"""
if not user.is_superuser(current_user):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions",
)
db_user = user.get(db, id=user_id)
if not db_user:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found",
)
updated_user = user.update(db, db_obj=db_user, obj_in=user_in)
return updated_user

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

@ -0,0 +1,46 @@
from typing import Generator
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from pydantic import ValidationError
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
from app.crud.crud_user import get_user_by_id
oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
def get_db() -> Generator:
"""
Dependency for getting database session.
"""
db = SessionLocal()
try:
yield db
finally:
db.close()
async def get_current_user(
db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> User:
"""
Dependency for getting the current authenticated user.
"""
try:
payload = jwt.decode(
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
)
token_data = TokenPayload(**payload)
except (JWTError, ValidationError):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Could not validate credentials",
)
user = get_user_by_id(db, id=token_data.sub)
if not user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
return user

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

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

@ -0,0 +1,37 @@
from typing import List, Union
from pathlib import Path
from pydantic import AnyHttpUrl, field_validator
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
API_V1_STR: str = "/api/v1"
PROJECT_NAME: str = "Small Business Inventory Management System"
# CORS Configuration
BACKEND_CORS_ORIGINS: List[AnyHttpUrl] = []
@field_validator("BACKEND_CORS_ORIGINS", mode="before")
def assemble_cors_origins(cls, v: Union[str, List[str]]) -> Union[List[str], str]:
if isinstance(v, str) and not v.startswith("["):
return [i.strip() for i in v.split(",")]
elif isinstance(v, (list, str)):
return v
raise ValueError(v)
# JWT Configuration
SECRET_KEY: str = "CHANGE_ME_IN_PRODUCTION" # Change this in a real production environment!
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8 # 8 days
# Database Configuration
DB_DIR: Path = Path("/app/storage/db")
DB_DIR.mkdir(parents=True, exist_ok=True)
SQLALCHEMY_DATABASE_URL: str = f"sqlite:///{DB_DIR}/db.sqlite"
class Config:
env_file = ".env"
case_sensitive = True
settings = Settings()

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

@ -0,0 +1,37 @@
from datetime import datetime, timedelta
from typing import Any, Union
from jose import jwt
from passlib.context import CryptContext
from app.core.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def create_access_token(
subject: Union[str, Any], expires_delta: timedelta = None
) -> str:
"""
Create a JWT access token.
"""
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
)
to_encode = {"exp": expire, "sub": str(subject)}
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
return encoded_jwt
def verify_password(plain_password: str, hashed_password: str) -> bool:
"""
Verify a password against a hash.
"""
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
"""
Generate a password hash.
"""
return pwd_context.hash(password)

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

@ -0,0 +1,6 @@
from app.crud.crud_user import user as user
from app.crud.crud_category import category as category
from app.crud.crud_supplier import supplier as supplier
from app.crud.crud_item import item as item, transaction as transaction
__all__ = ["user", "category", "supplier", "item", "transaction"]

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

@ -0,0 +1,66 @@
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.session import Base
ModelType = TypeVar("ModelType", bound=Base)
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
def __init__(self, model: Type[ModelType]):
"""
CRUD object with default methods to Create, Read, Update, Delete (CRUD).
**Parameters**
* `model`: A SQLAlchemy model class
* `schema`: A Pydantic model (schema) class
"""
self.model = model
def get(self, db: Session, id: Any) -> Optional[ModelType]:
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]:
return db.query(self.model).offset(skip).limit(limit).all()
def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:
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:
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:
obj = db.query(self.model).get(id)
db.delete(obj)
db.commit()
return obj

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

@ -0,0 +1,15 @@
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]):
def get_by_name(self, db: Session, *, name: str) -> Optional[Category]:
return db.query(Category).filter(Category.name == name).first()
category = CRUDCategory(Category)

97
app/crud/crud_item.py Normal file
View File

@ -0,0 +1,97 @@
from typing import List, Optional
from sqlalchemy.orm import Session
from app.crud.base import CRUDBase
from app.models.item import Item, Transaction, TransactionItem, TransactionType
from app.schemas.item import ItemCreate, ItemUpdate, TransactionCreate, TransactionUpdate, InventoryAdjustment
class CRUDItem(CRUDBase[Item, ItemCreate, ItemUpdate]):
def get_by_sku(self, db: Session, *, sku: str) -> Optional[Item]:
return db.query(Item).filter(Item.sku == sku).first()
def get_by_category(self, db: Session, *, category_id: int, skip: int = 0, limit: int = 100) -> List[Item]:
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]:
return db.query(Item).filter(Item.supplier_id == supplier_id).offset(skip).limit(limit).all()
def get_low_stock_items(self, db: Session, *, skip: int = 0, limit: int = 100) -> List[Item]:
return db.query(Item).filter(Item.quantity <= Item.reorder_level).offset(skip).limit(limit).all()
def adjust_inventory(self, db: Session, *, item_id: int, adjustment: InventoryAdjustment) -> Item:
item = self.get(db, id=item_id)
if not item:
return None
# Update item quantity
item.quantity += adjustment.quantity
# Create an adjustment transaction
transaction = Transaction(
transaction_type=TransactionType.ADJUSTMENT,
notes=adjustment.reason or f"Quantity adjusted by {adjustment.quantity}"
)
db.add(transaction)
db.flush()
# Create transaction item
transaction_item = TransactionItem(
transaction_id=transaction.id,
item_id=item_id,
quantity=abs(adjustment.quantity),
unit_price=item.unit_price
)
db.add(transaction_item)
db.commit()
db.refresh(item)
return item
class CRUDTransaction(CRUDBase[Transaction, TransactionCreate, TransactionUpdate]):
def create(self, db: Session, *, obj_in: TransactionCreate) -> Transaction:
# Create transaction
transaction = Transaction(
transaction_type=obj_in.transaction_type,
reference_number=obj_in.reference_number,
notes=obj_in.notes
)
db.add(transaction)
db.flush()
# Create transaction items
for item_data in obj_in.items:
transaction_item = TransactionItem(
transaction_id=transaction.id,
item_id=item_data.item_id,
quantity=item_data.quantity,
unit_price=item_data.unit_price
)
db.add(transaction_item)
# Update item quantity based on transaction type
item = db.query(Item).get(item_data.item_id)
if obj_in.transaction_type == TransactionType.PURCHASE:
item.quantity += item_data.quantity
elif obj_in.transaction_type == TransactionType.SALE:
item.quantity -= item_data.quantity
db.commit()
db.refresh(transaction)
return transaction
def get_transactions_by_item(self, db: Session, *, item_id: int, skip: int = 0, limit: int = 100) -> List[Transaction]:
return (
db.query(Transaction)
.join(TransactionItem)
.filter(TransactionItem.item_id == item_id)
.offset(skip)
.limit(limit)
.all()
)
item = CRUDItem(Item)
transaction = CRUDTransaction(Transaction)

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

@ -0,0 +1,15 @@
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]):
def get_by_name(self, db: Session, *, name: str) -> Optional[Supplier]:
return db.query(Supplier).filter(Supplier.name == name).first()
supplier = CRUDSupplier(Supplier)

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

@ -0,0 +1,71 @@
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]):
def get_by_email(self, db: Session, *, email: str) -> Optional[User]:
return db.query(User).filter(User.email == email).first()
def get_by_username(self, db: Session, *, username: str) -> Optional[User]:
return db.query(User).filter(User.username == username).first()
def create(self, db: Session, *, obj_in: UserCreate) -> User:
db_obj = User(
email=obj_in.email,
username=obj_in.username,
hashed_password=get_password_hash(obj_in.password),
full_name=obj_in.full_name,
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:
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]:
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:
return user.is_active
def is_superuser(self, user: User) -> bool:
return user.is_superuser
user = CRUDUser(User)
def get_user_by_id(db: Session, id: int) -> Optional[User]:
return user.get(db, id=id)
def get_user_by_email(db: Session, email: str) -> Optional[User]:
return user.get_by_email(db, email=email)
def get_user_by_username(db: Session, username: str) -> Optional[User]:
return user.get_by_username(db, username=username)

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

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

@ -0,0 +1,7 @@
# Import all the models, so that Base has them before being
# imported by Alembic
from app.db.session import Base # noqa
from app.models.user import User # noqa
from app.models.category import Category # noqa
from app.models.supplier import Supplier # noqa
from app.models.item import Item, Transaction, TransactionItem # noqa

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

@ -0,0 +1,18 @@
from pathlib import Path
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
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)
Base = declarative_base()

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

@ -0,0 +1,11 @@
from app.models.user import User as User
from app.models.category import Category as Category
from app.models.supplier import Supplier as Supplier
from app.models.item import (
Item as Item,
Transaction as Transaction,
TransactionItem as TransactionItem,
TransactionType as TransactionType
)
__all__ = ["User", "Category", "Supplier", "Item", "Transaction", "TransactionItem", "TransactionType"]

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

@ -0,0 +1,20 @@
from datetime import datetime
from sqlalchemy import Column, Integer, DateTime
from sqlalchemy.ext.declarative import declared_attr
from app.db.session import Base
class TimestampMixin:
"""Mixin that adds created_at and updated_at columns to a table."""
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)
class TableNameMixin:
"""Mixin that sets the table name to the lowercase pluralized class name."""
@declared_attr
def __tablename__(cls):
return cls.__name__.lower() + 's'
class BaseModel(TimestampMixin, TableNameMixin, Base):
"""Base model for all models to inherit from."""
__abstract__ = True
id = Column(Integer, primary_key=True, index=True)

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

@ -0,0 +1,11 @@
from sqlalchemy import Column, String, Text
from sqlalchemy.orm import relationship
from app.models.base import BaseModel
class Category(BaseModel):
"""Model for product categories."""
name = Column(String, unique=True, index=True, nullable=False)
description = Column(Text, nullable=True)
# Relationships
items = relationship("Item", back_populates="category")

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

@ -0,0 +1,48 @@
from sqlalchemy import Column, String, Integer, Float, ForeignKey, Text, Boolean
from sqlalchemy.orm import relationship
from app.models.base import BaseModel
class Item(BaseModel):
"""Model for inventory items."""
sku = Column(String, unique=True, index=True, nullable=False)
name = Column(String, index=True, nullable=False)
description = Column(Text, nullable=True)
quantity = Column(Integer, default=0, nullable=False)
reorder_level = Column(Integer, default=5, nullable=False)
unit_price = Column(Float, nullable=False)
category_id = Column(Integer, ForeignKey('categorys.id'), nullable=True)
supplier_id = Column(Integer, ForeignKey('suppliers.id'), nullable=True)
location = Column(String, nullable=True)
is_active = Column(Boolean, default=True)
# Relationships
category = relationship("Category", back_populates="items")
supplier = relationship("Supplier", back_populates="items")
transaction_items = relationship("TransactionItem", back_populates="item")
class TransactionType:
PURCHASE = "purchase"
SALE = "sale"
ADJUSTMENT = "adjustment"
RETURN = "return"
TRANSFER = "transfer"
class Transaction(BaseModel):
"""Model for inventory transactions (purchases, sales, adjustments, etc.)."""
transaction_type = Column(String, nullable=False)
reference_number = Column(String, nullable=True)
notes = Column(Text, nullable=True)
# Relationships
items = relationship("TransactionItem", back_populates="transaction")
class TransactionItem(BaseModel):
"""Model for items involved in a transaction."""
transaction_id = Column(Integer, ForeignKey('transactions.id'), nullable=False)
item_id = Column(Integer, ForeignKey('items.id'), nullable=False)
quantity = Column(Integer, nullable=False)
unit_price = Column(Float, nullable=False)
# Relationships
transaction = relationship("Transaction", back_populates="items")
item = relationship("Item", back_populates="transaction_items")

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

@ -0,0 +1,15 @@
from sqlalchemy import Column, String, Text
from sqlalchemy.orm import relationship
from app.models.base import BaseModel
class Supplier(BaseModel):
"""Model for suppliers."""
name = Column(String, unique=True, index=True, nullable=False)
contact_name = Column(String, nullable=True)
contact_email = Column(String, nullable=True)
contact_phone = Column(String, nullable=True)
address = Column(Text, nullable=True)
notes = Column(Text, nullable=True)
# Relationships
items = relationship("Item", back_populates="supplier")

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

@ -0,0 +1,11 @@
from sqlalchemy import Column, String, Boolean
from app.models.base import BaseModel
class User(BaseModel):
"""Model for user accounts."""
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, nullable=True)
is_active = Column(Boolean, default=True)
is_superuser = Column(Boolean, default=False)

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

@ -0,0 +1,21 @@
from app.schemas.token import Token, TokenPayload
from app.schemas.user import User, UserCreate, UserUpdate, UserInDB
from app.schemas.category import Category, CategoryCreate, CategoryUpdate
from app.schemas.supplier import Supplier, SupplierCreate, SupplierUpdate
from app.schemas.item import (
Item, ItemCreate, ItemUpdate,
Transaction, TransactionCreate, TransactionUpdate,
TransactionItem, TransactionItemCreate, TransactionItemUpdate,
InventoryAdjustment
)
__all__ = [
"Token", "TokenPayload",
"User", "UserCreate", "UserUpdate", "UserInDB",
"Category", "CategoryCreate", "CategoryUpdate",
"Supplier", "SupplierCreate", "SupplierUpdate",
"Item", "ItemCreate", "ItemUpdate",
"Transaction", "TransactionCreate", "TransactionUpdate",
"TransactionItem", "TransactionItemCreate", "TransactionItemUpdate",
"InventoryAdjustment"
]

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

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

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

@ -0,0 +1,93 @@
from typing import Optional, List
from pydantic import BaseModel, Field
class ItemBase(BaseModel):
sku: Optional[str] = None
name: Optional[str] = None
description: Optional[str] = None
quantity: Optional[int] = 0
reorder_level: Optional[int] = 5
unit_price: Optional[float] = None
category_id: Optional[int] = None
supplier_id: Optional[int] = None
location: Optional[str] = None
is_active: Optional[bool] = True
class ItemCreate(ItemBase):
sku: str
name: str
unit_price: float
class ItemUpdate(ItemBase):
pass
class ItemInDBBase(ItemBase):
id: int
class Config:
from_attributes = True
class Item(ItemInDBBase):
pass
class TransactionItemBase(BaseModel):
item_id: int
quantity: int
unit_price: float
class TransactionItemCreate(TransactionItemBase):
pass
class TransactionItemUpdate(TransactionItemBase):
pass
class TransactionItemInDBBase(TransactionItemBase):
id: int
transaction_id: int
class Config:
from_attributes = True
class TransactionItem(TransactionItemInDBBase):
pass
class TransactionBase(BaseModel):
transaction_type: str
reference_number: Optional[str] = None
notes: Optional[str] = None
class TransactionCreate(TransactionBase):
items: List[TransactionItemCreate]
class TransactionUpdate(TransactionBase):
pass
class TransactionInDBBase(TransactionBase):
id: int
class Config:
from_attributes = True
class Transaction(TransactionInDBBase):
items: List[TransactionItem] = []
class InventoryAdjustment(BaseModel):
item_id: int
quantity: int = Field(..., description="Positive for additions, negative for subtractions")
reason: Optional[str] = None

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

@ -0,0 +1,30 @@
from typing import Optional
from pydantic import BaseModel, EmailStr
class SupplierBase(BaseModel):
name: Optional[str] = None
contact_name: Optional[str] = None
contact_email: Optional[EmailStr] = None
contact_phone: Optional[str] = None
address: Optional[str] = None
notes: Optional[str] = None
class SupplierCreate(SupplierBase):
name: str
class SupplierUpdate(SupplierBase):
pass
class SupplierInDBBase(SupplierBase):
id: int
class Config:
from_attributes = True
class Supplier(SupplierInDBBase):
pass

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

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

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

@ -0,0 +1,35 @@
from typing import Optional
from pydantic import BaseModel, EmailStr
class UserBase(BaseModel):
username: Optional[str] = None
email: Optional[EmailStr] = None
full_name: Optional[str] = None
is_active: Optional[bool] = True
is_superuser: Optional[bool] = False
class UserCreate(UserBase):
username: str
email: EmailStr
password: str
class UserUpdate(UserBase):
password: Optional[str] = None
class UserInDBBase(UserBase):
id: int
class Config:
orm_mode = True
class User(UserInDBBase):
pass
class UserInDB(UserInDBBase):
hashed_password: str

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

37
main.py Normal file
View File

@ -0,0 +1,37 @@
import uvicorn
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.api_v1.api import api_router
from app.core.config import settings
app = FastAPI(
title=settings.PROJECT_NAME,
description="Small Business Inventory Management System API",
version="1.0.0",
openapi_url=f"{settings.API_V1_STR}/openapi.json",
docs_url="/docs",
redoc_url="/redoc",
)
# Set all CORS enabled origins
if settings.BACKEND_CORS_ORIGINS:
app.add_middleware(
CORSMiddleware,
allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(api_router, prefix=settings.API_V1_STR)
@app.get("/health", tags=["Health"])
async def health_check():
"""
Health check endpoint to verify the API is running.
"""
return {"status": "healthy"}
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

0
migrations/__init__.py Normal file
View File

88
migrations/env.py Normal file
View File

@ -0,0 +1,88 @@
import sys
from pathlib import Path
from logging.config import fileConfig
from sqlalchemy import engine_from_config
from sqlalchemy import pool
from alembic import context
# Add the parent directory to sys.path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
# Import models for Alembic autogenerate support
from app.db.base import Base # noqa
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
# Interpret the config file for Python logging.
# This line sets up loggers basically.
if config.config_file_name is not None:
fileConfig(config.config_file_name)
# add your model's MetaData object here
# for 'autogenerate' support
target_metadata = Base.metadata
# other values from the config, defined by the needs of env.py,
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode.
This configures the context with just a URL
and not an Engine, though an Engine is acceptable
here as well. By skipping the Engine creation
we don't even need a DBAPI to be available.
Calls to context.execute() here emit the given string to the
script output.
"""
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
render_as_batch=True,
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
"""Run migrations in 'online' mode.
In this scenario we need to create an Engine
and associate a connection with the context.
"""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
is_sqlite = connection.dialect.name == 'sqlite'
context.configure(
connection=connection,
target_metadata=target_metadata,
render_as_batch=is_sqlite,
)
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() -> None:
${upgrades if upgrades else "pass"}
def downgrade() -> None:
${downgrades if downgrades else "pass"}

128
migrations/versions/001_initial_migration.py Normal file
View File

@ -0,0 +1,128 @@
"""Initial migration
Revision ID: 001
Revises:
Create Date: 2023-12-14 00:00:00.000000
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = '001'
down_revision = None
branch_labels = None
depends_on = None
def upgrade() -> None:
# 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(), default=True),
sa.Column('is_superuser', sa.Boolean(), default=False),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False)
op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True)
# Create categories table
op.create_table(
'categorys', # Intentionally misspelled to match the model's table name
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_categorys_id'), 'categorys', ['id'], unique=False)
op.create_index(op.f('ix_categorys_name'), 'categorys', ['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('contact_email', sa.String(), nullable=True),
sa.Column('contact_phone', sa.String(), nullable=True),
sa.Column('address', sa.Text(), nullable=True),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_suppliers_id'), 'suppliers', ['id'], unique=False)
op.create_index(op.f('ix_suppliers_name'), 'suppliers', ['name'], unique=True)
# Create items table
op.create_table(
'items',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('sku', sa.String(), nullable=False),
sa.Column('name', sa.String(), nullable=False),
sa.Column('description', sa.Text(), nullable=True),
sa.Column('quantity', sa.Integer(), nullable=False, default=0),
sa.Column('reorder_level', sa.Integer(), nullable=False, default=5),
sa.Column('unit_price', sa.Float(), nullable=False),
sa.Column('category_id', sa.Integer(), nullable=True),
sa.Column('supplier_id', sa.Integer(), nullable=True),
sa.Column('location', sa.String(), nullable=True),
sa.Column('is_active', sa.Boolean(), default=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['category_id'], ['categorys.id']),
sa.ForeignKeyConstraint(['supplier_id'], ['suppliers.id']),
sa.PrimaryKeyConstraint('id'),
)
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('transaction_type', sa.String(), nullable=False),
sa.Column('reference_number', sa.String(), nullable=True),
sa.Column('notes', sa.Text(), nullable=True),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_transactions_id'), 'transactions', ['id'], unique=False)
# Create transaction_items table
op.create_table(
'transactionitems',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('transaction_id', sa.Integer(), nullable=False),
sa.Column('item_id', sa.Integer(), nullable=False),
sa.Column('quantity', sa.Integer(), nullable=False),
sa.Column('unit_price', sa.Float(), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.Column('updated_at', sa.DateTime(), nullable=False),
sa.ForeignKeyConstraint(['transaction_id'], ['transactions.id']),
sa.ForeignKeyConstraint(['item_id'], ['items.id']),
sa.PrimaryKeyConstraint('id'),
)
op.create_index(op.f('ix_transactionitems_id'), 'transactionitems', ['id'], unique=False)
def downgrade() -> None:
op.drop_table('transactionitems')
op.drop_table('transactions')
op.drop_table('items')
op.drop_table('suppliers')
op.drop_table('categorys')
op.drop_table('users')

0
migrations/versions/__init__.py Normal file
View File

10
requirements.txt Normal file
View File

@ -0,0 +1,10 @@
fastapi>=0.100.0
uvicorn>=0.23.0
sqlalchemy>=2.0.0
alembic>=1.12.0
pydantic>=2.0.0
python-multipart>=0.0.6
python-jose[cryptography]>=3.3.0
passlib[bcrypt]>=1.7.4
ruff>=0.1.0
python-dotenv>=1.0.0