4dc182713c
- Created project structure with FastAPI framework - Set up SQLite database with SQLAlchemy ORM - Implemented models: User, Item, Supplier, Transaction - Created CRUD operations for all models - Added API endpoints for all resources - Implemented JWT authentication and authorization - Added Alembic migration scripts - Created health endpoint - Updated documentation generated with BackendIM... (backend.im)
60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
from typing import List, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.crud.crud_item import item as crud_item
|
|
from app.models.transaction import Transaction, TransactionType
|
|
from app.schemas.transaction import TransactionCreate, TransactionUpdate
|
|
|
|
|
|
class CRUDTransaction(CRUDBase[Transaction, TransactionCreate, TransactionUpdate]):
|
|
def create(self, db: Session, *, obj_in: TransactionCreate) -> Transaction:
|
|
# Create transaction
|
|
db_obj = Transaction(
|
|
transaction_type=obj_in.transaction_type,
|
|
quantity=obj_in.quantity,
|
|
price_per_unit=obj_in.price_per_unit,
|
|
reference_number=obj_in.reference_number,
|
|
notes=obj_in.notes,
|
|
item_id=obj_in.item_id,
|
|
user_id=obj_in.user_id,
|
|
transaction_date=obj_in.transaction_date
|
|
)
|
|
db.add(db_obj)
|
|
|
|
# Update item quantity based on transaction type
|
|
quantity_change = obj_in.quantity
|
|
if obj_in.transaction_type in [TransactionType.SALE, TransactionType.RETURN]:
|
|
quantity_change = -quantity_change
|
|
|
|
crud_item.update_quantity(db, item_id=obj_in.item_id, quantity_change=quantity_change)
|
|
|
|
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]:
|
|
return (
|
|
db.query(Transaction)
|
|
.filter(Transaction.item_id == item_id)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_by_transaction_type(
|
|
self, db: Session, *, transaction_type: TransactionType, skip: int = 0, limit: int = 100
|
|
) -> List[Transaction]:
|
|
return (
|
|
db.query(Transaction)
|
|
.filter(Transaction.transaction_type == transaction_type)
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
|
|
transaction = CRUDTransaction(Transaction) |