700da98f88
- Created complete RESTful API for inventory management - Set up database models for items, categories, suppliers, and transactions - Implemented user authentication with JWT tokens - Added transaction tracking for inventory movements - Created comprehensive API endpoints for all CRUD operations - Set up Alembic for database migrations - Added input validation and error handling - Created detailed documentation in README
47 lines
2.1 KiB
Python
47 lines
2.1 KiB
Python
from typing import List
|
|
import uuid
|
|
from datetime import datetime
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.transaction import Transaction
|
|
from app.schemas.transaction import TransactionCreate, TransactionUpdate
|
|
|
|
|
|
class CRUDTransaction(CRUDBase[Transaction, TransactionCreate, TransactionUpdate]):
|
|
def create(self, db: Session, *, obj_in: TransactionCreate) -> Transaction:
|
|
transaction_date = obj_in.transaction_date or datetime.now()
|
|
db_obj = Transaction(
|
|
id=str(uuid.uuid4()),
|
|
transaction_type=obj_in.transaction_type,
|
|
quantity=obj_in.quantity,
|
|
unit_price=obj_in.unit_price,
|
|
total_price=obj_in.total_price,
|
|
reference_number=obj_in.reference_number,
|
|
notes=obj_in.notes,
|
|
transaction_date=transaction_date,
|
|
item_id=obj_in.item_id,
|
|
user_id=obj_in.user_id,
|
|
)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def get_by_item(self, db: Session, *, item_id: str, 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_user(self, db: Session, *, user_id: str, skip: int = 0, limit: int = 100) -> List[Transaction]:
|
|
return db.query(Transaction).filter(Transaction.user_id == user_id).offset(skip).limit(limit).all()
|
|
|
|
def get_by_type(self, db: Session, *, transaction_type: str, skip: int = 0, limit: int = 100) -> List[Transaction]:
|
|
return db.query(Transaction).filter(Transaction.transaction_type == transaction_type).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]:
|
|
return db.query(Transaction).filter(
|
|
Transaction.transaction_date >= start_date,
|
|
Transaction.transaction_date <= end_date
|
|
).offset(skip).limit(limit).all()
|
|
|
|
|
|
transaction = CRUDTransaction(Transaction) |