Cheng Zhou
75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
import uuid
|
|
from typing import Sequence
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.contract_fee_payment import ContractFeePayment
|
|
from app.schemas.contract_fee_payment import ContractFeePaymentCreate, ContractFeePaymentUpdate
|
|
|
|
|
|
async def create_payment(
|
|
db: AsyncSession,
|
|
contract_fee_id: uuid.UUID,
|
|
payment_in: ContractFeePaymentCreate,
|
|
) -> ContractFeePayment:
|
|
result = await db.execute(
|
|
select(func.coalesce(func.max(ContractFeePayment.seq), 0)).where(
|
|
ContractFeePayment.contract_fee_id == contract_fee_id
|
|
)
|
|
)
|
|
next_seq = (result.scalar_one_or_none() or 0) + 1
|
|
payment = ContractFeePayment(
|
|
contract_fee_id=contract_fee_id,
|
|
seq=next_seq,
|
|
amount=payment_in.amount,
|
|
paid_date=payment_in.paid_date,
|
|
verified_date=payment_in.verified_date,
|
|
is_paid=payment_in.is_paid,
|
|
is_verified=payment_in.is_verified,
|
|
remark=payment_in.remark,
|
|
)
|
|
db.add(payment)
|
|
await db.commit()
|
|
await db.refresh(payment)
|
|
return payment
|
|
|
|
|
|
async def get_payment(db: AsyncSession, payment_id: uuid.UUID) -> ContractFeePayment | None:
|
|
result = await db.execute(select(ContractFeePayment).where(ContractFeePayment.id == payment_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_payments(db: AsyncSession, contract_fee_id: uuid.UUID) -> Sequence[ContractFeePayment]:
|
|
result = await db.execute(
|
|
select(ContractFeePayment)
|
|
.where(ContractFeePayment.contract_fee_id == contract_fee_id)
|
|
.order_by(ContractFeePayment.seq.asc(), ContractFeePayment.created_at.asc())
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def update_payment(
|
|
db: AsyncSession, payment: ContractFeePayment, payment_in: ContractFeePaymentUpdate
|
|
) -> ContractFeePayment:
|
|
update_data = payment_in.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(payment, key, value)
|
|
await db.commit()
|
|
await db.refresh(payment)
|
|
return payment
|
|
|
|
|
|
async def delete_payment(db: AsyncSession, payment: ContractFeePayment) -> None:
|
|
await db.delete(payment)
|
|
await db.commit()
|
|
|
|
|
|
async def resequence_payments(db: AsyncSession, contract_fee_id: uuid.UUID) -> None:
|
|
payments = await list_payments(db, contract_fee_id)
|
|
for idx, payment in enumerate(payments, start=1):
|
|
if payment.seq != idx:
|
|
payment.seq = idx
|
|
db.add(payment)
|
|
await db.commit()
|