54 lines
1.5 KiB
Python
54 lines
1.5 KiB
Python
from datetime import date, datetime
|
|
from typing import List
|
|
|
|
from fastapi import APIRouter, Security
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..core.security import get_current_active_user as get_user
|
|
from ..db.session import SessionFuture
|
|
from ..models.voucher import Voucher
|
|
from ..models.voucher_type import VoucherType
|
|
from ..schemas.issue_grid_item import IssueGridItem
|
|
from ..schemas.user import UserToken
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.get("/{date_}", response_model=List[IssueGridItem])
|
|
def grid_date(
|
|
date_: str,
|
|
user: UserToken = Security(get_user, scopes=["issue"]),
|
|
) -> List[IssueGridItem]:
|
|
date_obj = datetime.strptime(date_, "%d-%b-%Y")
|
|
with SessionFuture() as db:
|
|
return get_grid(date_obj, db)
|
|
|
|
|
|
def get_grid(date_: date, db: Session) -> List[IssueGridItem]:
|
|
list_: List[IssueGridItem] = []
|
|
query = (
|
|
db.execute(
|
|
select(Voucher)
|
|
.join(Voucher.journals)
|
|
.where(Voucher.date == date_, Voucher.voucher_type == VoucherType.ISSUE)
|
|
.order_by(Voucher.creation_date)
|
|
)
|
|
.unique()
|
|
.scalars()
|
|
.all()
|
|
)
|
|
|
|
for voucher in query:
|
|
amount = voucher.journals[0].amount
|
|
list_.append(
|
|
IssueGridItem(
|
|
id=str(voucher.id),
|
|
amount=amount,
|
|
source=next(j.cost_centre.name for j in voucher.journals if j.debit == -1),
|
|
destination=next(j.cost_centre.name for j in voucher.journals if j.debit == 1),
|
|
)
|
|
)
|
|
return list_
|