59 lines
1.6 KiB
Python
59 lines
1.6 KiB
Python
from typing import List, Optional
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.crud.base import CRUDBase
|
|
from app.models.kyc import KYC, KYCStatus
|
|
from app.schemas.kyc import KYCCreate, KYCUpdate
|
|
|
|
|
|
class CRUDKYC(CRUDBase[KYC, KYCCreate, KYCUpdate]):
|
|
def get_by_user(
|
|
self, db: Session, *, user_id: int
|
|
) -> Optional[KYC]:
|
|
return db.query(KYC).filter(KYC.user_id == user_id).first()
|
|
|
|
def get_by_status(
|
|
self, db: Session, *, status: str, skip: int = 0, limit: int = 100
|
|
) -> List[KYC]:
|
|
return (
|
|
db.query(KYC)
|
|
.filter(KYC.status == status)
|
|
.order_by(KYC.created_at.desc())
|
|
.offset(skip)
|
|
.limit(limit)
|
|
.all()
|
|
)
|
|
|
|
def get_all_pending(
|
|
self, db: Session, *, skip: int = 0, limit: int = 100
|
|
) -> List[KYC]:
|
|
return self.get_by_status(db, status=KYCStatus.PENDING, skip=skip, limit=limit)
|
|
|
|
def approve(
|
|
self, db: Session, *, db_obj: KYC
|
|
) -> KYC:
|
|
update_data = {"status": KYCStatus.APPROVED}
|
|
return super().update(db, db_obj=db_obj, obj_in=update_data)
|
|
|
|
def reject(
|
|
self, db: Session, *, db_obj: KYC, rejection_reason: str
|
|
) -> KYC:
|
|
update_data = {
|
|
"status": KYCStatus.REJECTED,
|
|
"rejection_reason": rejection_reason
|
|
}
|
|
return super().update(db, db_obj=db_obj, obj_in=update_data)
|
|
|
|
|
|
kyc = CRUDKYC(KYC)
|
|
|
|
|
|
# Aliases for convenience
|
|
get_kyc = kyc.get
|
|
create_kyc = kyc.create
|
|
get_kyc_by_user = kyc.get_by_user
|
|
get_kyc_by_status = kyc.get_by_status
|
|
get_all_pending_kyc = kyc.get_all_pending
|
|
approve_kyc = kyc.approve
|
|
reject_kyc = kyc.reject |