250 lines
9.9 KiB
Python
250 lines
9.9 KiB
Python
import uuid
|
|
|
|
from typing import List
|
|
|
|
import luthor.schemas.case as schemas
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Security, status
|
|
from sqlalchemy import desc
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from ..core.security import get_current_active_user as get_user
|
|
from ..db.session import SessionLocal
|
|
from ..models.case import Case
|
|
from ..schemas.user_token import UserToken
|
|
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# Dependency
|
|
def get_db():
|
|
try:
|
|
db = SessionLocal()
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
@router.post("", response_model=schemas.Case)
|
|
def save(
|
|
data: schemas.CaseIn,
|
|
db: Session = Depends(get_db),
|
|
user: UserToken = Security(get_user, scopes=["cases"]),
|
|
) -> schemas.Case:
|
|
try:
|
|
item = Case(
|
|
office_file_number=data.office_file_number,
|
|
court_case_number=data.court_case_number,
|
|
year=data.year,
|
|
title=data.title,
|
|
docket_number=data.docket_number,
|
|
receipt_date=data.receipt_date,
|
|
limitation_date=data.limitation_date,
|
|
filing_date=data.filing_date,
|
|
appear_on_behalf_of=data.appear_on_behalf_of,
|
|
question_of_law=data.question_of_law,
|
|
aor_name=data.aor_name,
|
|
opposing_council_aor=data.opposing_council_aor,
|
|
opposing_council_detail=data.opposing_council_detail,
|
|
lower_court_case_number=data.lower_court_case_number,
|
|
date_of_impugned_judgement=data.date_of_impugned_judgement,
|
|
brief_description=data.brief_description,
|
|
remarks=data.remarks,
|
|
slp_counter=data.slp_counter,
|
|
contact_detail=data.contact_detail,
|
|
case_connected_with=data.case_connected_with,
|
|
bunch_cases=data.bunch_cases,
|
|
court_id=data.court.id_ if data.court is not None else None,
|
|
department_id=data.department.id_ if data.department is not None else None,
|
|
office_id=data.office.id_ if data.office is not None else None,
|
|
case_type_id=data.case_type.id_ if data.case_type is not None else None,
|
|
act_id=data.act.id_ if data.act is not None else None,
|
|
nature_id=data.nature.id_ if data.nature is not None else None,
|
|
office_status_id=data.office_status.id_ if data.office_status is not None else None,
|
|
court_status_id=data.court_status.id_ if data.court_status is not None else None,
|
|
)
|
|
db.add(item)
|
|
db.commit()
|
|
return case_info(item)
|
|
except SQLAlchemyError as e:
|
|
db.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
|
|
|
|
@router.put("/{id_}", response_model=schemas.Case)
|
|
def update(
|
|
id_: uuid.UUID,
|
|
data: schemas.CaseIn,
|
|
db: Session = Depends(get_db),
|
|
user: UserToken = Security(get_user, scopes=["cases"]),
|
|
) -> schemas.Case:
|
|
try:
|
|
item: Case = db.query(Case).filter(Case.id == id_).first()
|
|
item.office_file_number = data.office_file_number
|
|
item.court_case_number = data.court_case_number
|
|
item.year = data.year
|
|
item.title = data.title
|
|
item.docket_number = data.docket_number
|
|
item.receipt_date = data.receipt_date
|
|
item.limitation_date = data.limitation_date
|
|
item.filing_date = data.filing_date
|
|
item.appear_on_behalf_of = data.appear_on_behalf_of
|
|
item.question_of_law = data.question_of_law
|
|
item.aor_name = data.aor_name
|
|
item.opposing_council_aor = data.opposing_council_aor
|
|
item.opposing_council_detail = data.opposing_council_detail
|
|
item.lower_court_case_number = data.lower_court_case_number
|
|
item.date_of_impugned_judgement = data.date_of_impugned_judgement
|
|
item.brief_description = data.brief_description
|
|
item.remarks = data.remarks
|
|
item.slp_counter = data.slp_counter
|
|
item.contact_detail = data.contact_detail
|
|
item.case_connected_with = data.case_connected_with
|
|
item.bunch_cases = data.bunch_cases
|
|
item.court_id = data.court.id_ if data.court is not None else None
|
|
item.department_id = data.department.id_ if data.department is not None else None
|
|
item.office_id = data.office.id_ if data.office is not None else None
|
|
item.case_type_id = data.case_type.id_ if data.case_type is not None else None
|
|
item.act_id = data.act.id_ if data.act is not None else None
|
|
item.nature_id = data.nature.id_ if data.nature is not None else None
|
|
item.office_status_id = data.office_status.id_ if data.office_status is not None else None
|
|
item.court_status_id = data.court_status.id_ if data.court_status is not None else None
|
|
db.commit()
|
|
return case_info(item)
|
|
except SQLAlchemyError as e:
|
|
db.rollback()
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail=str(e),
|
|
)
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
|
|
|
|
@router.delete("/{id_}", response_model=schemas.CaseBlank)
|
|
def delete(
|
|
id_: uuid.UUID,
|
|
db: Session = Depends(get_db),
|
|
user: UserToken = Security(get_user, scopes=["cases"]),
|
|
) -> schemas.CaseBlank:
|
|
try:
|
|
item: Case = db.query(Case).filter(Case.id == id_).first()
|
|
if item is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Sale Category not found",
|
|
)
|
|
item.is_deleted = True
|
|
db.commit()
|
|
return case_blank()
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
|
|
|
|
@router.get("", response_model=schemas.CaseBlank)
|
|
def show_blank(
|
|
db: Session = Depends(get_db),
|
|
user: UserToken = Security(get_user, scopes=["cases"]),
|
|
) -> schemas.CaseBlank:
|
|
return case_blank()
|
|
|
|
|
|
@router.get("/list", response_model=List[schemas.Case])
|
|
def show_list(db: Session = Depends(get_db), user: UserToken = Depends(get_user)) -> List[schemas.Case]:
|
|
return [case_info(item) for item in db.query(Case).order_by(desc(Case.receipt_date)).all()]
|
|
|
|
|
|
@router.get("/list/{source_id}", response_model=List[schemas.Case])
|
|
def show_list_of_source(
|
|
source_id: uuid.UUID, db: Session = Depends(get_db), user: UserToken = Depends(get_user)
|
|
) -> List[schemas.Case]:
|
|
return [
|
|
case_info(item)
|
|
for item in db.query(Case).filter(Case.case_source_id == source_id).order_by(desc(Case.receipt_date)).all()
|
|
]
|
|
|
|
|
|
@router.get("/{id_}", response_model=schemas.Case)
|
|
def show_id(
|
|
id_: uuid.UUID,
|
|
db: Session = Depends(get_db),
|
|
user: UserToken = Security(get_user, scopes=["cases"]),
|
|
) -> schemas.Case:
|
|
item: Case = db.query(Case).filter(Case.id == id_).first()
|
|
return case_info(item)
|
|
|
|
|
|
def case_info(item: Case) -> schemas.Case:
|
|
return schemas.Case(
|
|
id=item.id,
|
|
officeFileNumber=item.office_file_number if item.office_file_number is not None else "",
|
|
courtCaseNumber=item.court_case_number if item.court_case_number is not None else "",
|
|
year=item.year,
|
|
title=item.title if item.title is not None else "",
|
|
docketNumber=item.docket_number if item.docket_number is not None else "",
|
|
receiptDate=item.receipt_date,
|
|
limitation_date=item.limitation_date,
|
|
filingDate=item.filing_date,
|
|
appearOnBehalfOf=item.appear_on_behalf_of if item.appear_on_behalf_of is not None else "",
|
|
questionOfLaw=item.question_of_law if item.question_of_law is not None else "",
|
|
aorName=item.aor_name if item.aor_name is not None else "",
|
|
opposingCouncilAor=item.opposing_council_aor if item.opposing_council_aor is not None else "",
|
|
opposingCouncilDetail=item.opposing_council_detail if item.opposing_council_detail is not None else "",
|
|
lowerCourtCaseNumber=item.lower_court_case_number if item.lower_court_case_number is not None else "",
|
|
dateOfImpugnedJudgement=item.date_of_impugned_judgement,
|
|
briefDescription=item.brief_description if item.brief_description is not None else "",
|
|
remarks=item.remarks if item.remarks is not None else "",
|
|
slpCounter=item.slp_counter if item.slp_counter is not None else "",
|
|
contactDetail=item.contact_detail if item.contact_detail is not None else "",
|
|
caseConnectedWith=item.case_connected_with if item.case_connected_with is not None else "",
|
|
bunchCases=item.bunch_cases if item.bunch_cases is not None else "",
|
|
court=schemas.CourtLink(id=item.court.id, name=item.court.name) if item.court is not None else None,
|
|
department=schemas.DepartmentLink(id=item.department.id, name=item.department.name)
|
|
if item.department is not None
|
|
else None,
|
|
office=schemas.OfficeLink(id=item.office.id, name=item.office.name) if item.office is not None else None,
|
|
caseType=schemas.CaseTypeLink(id=item.case_type.id, name=item.case_type.name)
|
|
if item.case_type is not None
|
|
else None,
|
|
act=schemas.ActLink(id=item.act.id, name=item.act.name) if item.act is not None else None,
|
|
nature=schemas.NatureLink(id=item.nature.id, name=item.nature.name) if item.nature is not None else None,
|
|
officeStatus=schemas.OfficeStatusLink(id=item.office_status.id, name=item.office_status.name)
|
|
if item.office_status is not None
|
|
else None,
|
|
courtStatus=schemas.CourtStatusLink(id=item.court_status.id, name=item.court_status.name)
|
|
if item.court_status is not None
|
|
else None,
|
|
)
|
|
|
|
|
|
def case_blank() -> schemas.CaseBlank:
|
|
return schemas.CaseBlank(
|
|
officeFileNumber="",
|
|
courtCaseNumber="",
|
|
year="",
|
|
title="",
|
|
docketNumber="",
|
|
appearOnBehalfOf="",
|
|
questionOfLaw="",
|
|
aorName="",
|
|
opposingCouncilAor="",
|
|
opposingCouncilDetail="",
|
|
lowerCourtCaseNumber="",
|
|
briefDescription="",
|
|
remarks="",
|
|
slpCounter="",
|
|
contactDetail="",
|
|
caseConnectedWith="",
|
|
bunchCases="",
|
|
)
|