ab-ms-core / app /controllers /reference.py
rajeshg820's picture
feat(reference): add customer_type_id filtering to reference data endpoints
f9a1cab
from fastapi import APIRouter, Depends, Query, status, HTTPException, Response
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.services.reference_service import ReferenceDataService
from app.schemas.reference import (
StateOut, CountryOut, CompanyTypeOut, LeadSourceOut, PaymentTermOut,
PurchasePriceOut, RentalPriceOut, BarrierSizeOut, ProductApplicationOut,
ReferenceDataResponse, FOBOut, EstShipDateOut, EstFreightOut, StatusInfoOut, ProjectReferenceResponse, PriorityOut,
StateCreateIn, StateUpdateIn, CountryCreateIn, CountryUpdateIn, CompanyTypeCreateIn, CompanyTypeUpdateIn,
LeadSourceCreateIn, LeadSourceUpdateIn, PaymentTermCreateIn, PaymentTermUpdateIn,
PurchasePriceCreateIn, PurchasePriceUpdateIn, RentalPriceCreateIn, RentalPriceUpdateIn,
BarrierSizeCreateIn, BarrierSizeUpdateIn, ProductApplicationCreateIn, ProductApplicationUpdateIn,
FOBCreateIn, FOBUpdateIn, EstShipDateCreateIn, EstShipDateUpdateIn,
EstFreightCreateIn, EstFreightUpdateIn, StatusInfoCreateIn, StatusInfoUpdateIn,
PriorityCreateIn, PriorityUpdateIn, PhoneTypeOut, PhoneTypeCreateIn, PhoneTypeUpdateIn
)
from app.core.dependencies import get_current_user_optional
from app.schemas.auth import CurrentUser
from typing import List, Optional
import logging
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/v1/reference", tags=["reference-data"])
@router.get("/all", response_model=ReferenceDataResponse)
def get_all_reference_data(
active_only: bool = Query(True, description="Return only active records"),
customer_type_id: Optional[int] = Query(None, description="Filter by customer type ID"),
db: Session = Depends(get_db),
current_user: Optional[CurrentUser] = Depends(get_current_user_optional)
):
"""
Get all reference/lookup data in a single response.
This endpoint is designed for frontend applications that need to populate
all dropdowns and lookup data efficiently. It includes:
- States/Provinces
- Countries
- Company Types
- Lead Sources
- Payment Terms
- Purchase Prices
- Rental Prices
- Barrier Sizes
- Product Applications
- Project Statuses
**Note**: This endpoint is public but may have enhanced data for authenticated users.
"""
service = ReferenceDataService(db)
return service.get_all_reference_data(active_only=active_only, customer_type_id=customer_type_id)
@router.get("/states", response_model=List[StateOut])
def get_states(
active_only: bool = Query(True, description="Return only active states"),
customer_type_id: Optional[int] = Query(None, description="Filter by customer type ID"),
db: Session = Depends(get_db)
):
"""
Get all states/provinces.
Returns a list of all states and provinces available in the system.
"""
service = ReferenceDataService(db)
return service.get_states(active_only=active_only, customer_type_id=customer_type_id)
@router.get("/countries", response_model=List[CountryOut])
def get_countries(
active_only: bool = Query(True, description="Return only active countries"),
db: Session = Depends(get_db)
):
"""
Get all countries.
Returns a list of all countries available in the system.
"""
service = ReferenceDataService(db)
return service.get_countries(active_only=active_only)
@router.get("/company-types", response_model=List[CompanyTypeOut])
def get_company_types(
active_only: bool = Query(True, description="Return only active company types"),
customer_type_id: Optional[int] = Query(None, description="Filter by customer type ID"),
db: Session = Depends(get_db)
):
"""
Get all company types.
Returns a list of all company types used for customer classification.
Optionally filtered by customer_type_id.
"""
service = ReferenceDataService(db)
return service.get_company_types(active_only=active_only, customer_type_id=customer_type_id)
@router.get("/lead-sources", response_model=List[LeadSourceOut])
def get_lead_sources(
active_only: bool = Query(True, description="Return only active lead sources"),
customer_type_id: Optional[int] = Query(None, description="Filter by customer type ID"),
db: Session = Depends(get_db)
):
"""
Get all lead generation sources.
Returns a list of all sources where leads can be generated from.
Optionally filtered by customer_type_id.
"""
service = ReferenceDataService(db)
return service.get_lead_sources(active_only=active_only, customer_type_id=customer_type_id)
@router.get("/payment-terms", response_model=List[PaymentTermOut])
def get_payment_terms(
active_only: bool = Query(True, description="Return only active payment terms"),
db: Session = Depends(get_db)
):
"""
Get all payment terms.
Returns a list of all available payment terms for transactions.
"""
service = ReferenceDataService(db)
return service.get_payment_terms(active_only=active_only)
@router.get("/purchase-prices", response_model=List[PurchasePriceOut])
def get_purchase_prices(
active_only: bool = Query(True, description="Return only active purchase prices"),
db: Session = Depends(get_db)
):
"""
Get all purchase prices.
Returns a list of all available purchase price options.
"""
service = ReferenceDataService(db)
return service.get_purchase_prices(active_only=active_only)
@router.get("/rental-prices", response_model=List[RentalPriceOut])
def get_rental_prices(
active_only: bool = Query(True, description="Return only active rental prices"),
db: Session = Depends(get_db)
):
"""
Get all rental prices.
Returns a list of all available rental price options.
"""
service = ReferenceDataService(db)
return service.get_rental_prices(active_only=active_only)
@router.get("/product-applications", response_model=List[ProductApplicationOut])
def get_product_applications(
active_only: bool = Query(True, description="Return only active product applications"),
db: Session = Depends(get_db)
):
"""
Get all product applications.
Returns a list of all available product application types.
"""
service = ReferenceDataService(db)
return service.get_product_applications(active_only=active_only)
@router.get("/fobs", response_model=List[FOBOut])
def get_FOB(
active_only: bool = Query(True, description="Return only active FOBS"),
db: Session = Depends(get_db)
):
"""
Get all FOBs (Free on Board terms).
Returns a list of all available FOB terms used in shipping.
"""
service = ReferenceDataService(db)
return service.get_fobs(active_only=active_only)
@router.get("/est-ship-dates", response_model=List[EstShipDateOut])
def get_est_ship_dates(
active_only: bool = Query(True, description="Return only active estimated ship dates"),
db: Session = Depends(get_db)
):
"""
Get all Estimated Ship Date options.
Returns a list of estimated shipping date options from the DB.
"""
service = ReferenceDataService(db)
return service.get_est_ship_dates(active_only=active_only)
@router.get("/est-freights", response_model=List[EstFreightOut])
def get_est_freights(
active_only: bool = Query(True, description="Return only active estimated freights"),
db: Session = Depends(get_db)
):
"""
Get all Estimated Freight options.
Returns a list of estimated freight options from the DB.
"""
service = ReferenceDataService(db)
return service.get_est_freights(active_only=active_only)
@router.get("/status-info", response_model=List[StatusInfoOut])
def get_status_info(
active_only: bool = Query(True, description="Return only active status info records"),
db: Session = Depends(get_db)
):
"""
Get all StatusInfo records.
Returns a list of project status metadata (ID, description, abbreviation).
"""
service = ReferenceDataService(db)
return service.get_status_info(active_only=active_only)
@router.get("/project-reference", response_model=ProjectReferenceResponse)
def get_project_reference(
active_only: bool = Query(True, description="Return only active project reference records"),
db: Session = Depends(get_db)
):
"""
Return grouped reference data used for project creation forms.
Only returns the subset of reference tables needed by project forms to reduce payload.
"""
service = ReferenceDataService(db)
return service.get_project_reference(active_only=active_only)
@router.post("/clear-cache", status_code=status.HTTP_200_OK)
def clear_reference_cache(
current_user: CurrentUser = Depends(get_current_user_optional),
db: Session = Depends(get_db)
):
"""
Clear the reference data cache.
Forces fresh data to be loaded on the next request.
This endpoint may require authentication in production.
"""
service = ReferenceDataService(db)
service.clear_cache()
return {"message": "Reference data cache cleared successfully"}
@router.get("/priorities", response_model=List[PriorityOut])
def get_priorities(
active_only: bool = Query(True, description="Return only active priorities"),
customer_type_id: Optional[int] = Query(None, description="Filter by customer type ID"),
db: Session = Depends(get_db)
):
"""
Get all priorities.
Returns a list of all available priorities, optionally filtered by customer type.
"""
service = ReferenceDataService(db)
return service.get_priorities(active_only=active_only, customer_type_id=customer_type_id)
@router.post("/states", response_model=StateOut, status_code=status.HTTP_201_CREATED)
def create_state(
data: StateCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new state/province.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_state(data)
@router.put("/states/{state_id}", response_model=StateOut)
def update_state(
state_id: int,
data: StateUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing state/province.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_state(state_id, data)
if not result:
raise HTTPException(status_code=404, detail="State not found")
return result
@router.post("/countries", response_model=CountryOut, status_code=status.HTTP_201_CREATED)
def create_country(
data: CountryCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new country.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_country(data)
@router.put("/countries/{country_id}", response_model=CountryOut)
def update_country(
country_id: int,
data: CountryUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing country.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_country(country_id, data)
if not result:
raise HTTPException(status_code=404, detail="Country not found")
return result
@router.post("/company-types", response_model=CompanyTypeOut, status_code=status.HTTP_201_CREATED)
def create_company_type(
data: CompanyTypeCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new company type.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_company_type(data)
@router.put("/company-types/{company_type_id}", response_model=CompanyTypeOut)
def update_company_type(
company_type_id: int,
data: CompanyTypeUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing company type.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_company_type(company_type_id, data)
if not result:
raise HTTPException(status_code=404, detail="Company type not found")
return result
@router.post("/lead-sources", response_model=LeadSourceOut, status_code=status.HTTP_201_CREATED)
def create_lead_source(
data: LeadSourceCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new lead source.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_lead_source(data)
@router.put("/lead-sources/{lead_generated_from_id}", response_model=LeadSourceOut)
def update_lead_source(
lead_generated_from_id: int,
data: LeadSourceUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing lead source.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_lead_source(lead_generated_from_id, data)
if not result:
raise HTTPException(status_code=404, detail="Lead source not found")
return result
@router.post("/payment-terms", response_model=PaymentTermOut, status_code=status.HTTP_201_CREATED)
def create_payment_term(
data: PaymentTermCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new payment term.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_payment_term(data)
@router.put("/payment-terms/{payment_term_id}", response_model=PaymentTermOut)
def update_payment_term(
payment_term_id: int,
data: PaymentTermUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing payment term.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_payment_term(payment_term_id, data)
if not result:
raise HTTPException(status_code=404, detail="Payment term not found")
return result
@router.post("/purchase-prices", response_model=PurchasePriceOut, status_code=status.HTTP_201_CREATED)
def create_purchase_price(
data: PurchasePriceCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new purchase price.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_purchase_price(data)
@router.put("/purchase-prices/{purchase_price_id}", response_model=PurchasePriceOut)
def update_purchase_price(
purchase_price_id: int,
data: PurchasePriceUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing purchase price.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_purchase_price(purchase_price_id, data)
if not result:
raise HTTPException(status_code=404, detail="Purchase price not found")
return result
@router.post("/rental-prices", response_model=RentalPriceOut, status_code=status.HTTP_201_CREATED)
def create_rental_price(
data: RentalPriceCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new rental price.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_rental_price(data)
@router.put("/rental-prices/{rental_price_id}", response_model=RentalPriceOut)
def update_rental_price(
rental_price_id: int,
data: RentalPriceUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing rental price.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_rental_price(rental_price_id, data)
if not result:
raise HTTPException(status_code=404, detail="Rental price not found")
return result
@router.post("/barrier-sizes", response_model=BarrierSizeOut, status_code=status.HTTP_201_CREATED)
def create_barrier_size(
data: BarrierSizeCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new barrier size.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_barrier_size(data)
@router.put("/barrier-sizes/{barrier_size_id}", response_model=BarrierSizeOut)
def update_barrier_size(
barrier_size_id: int,
data: BarrierSizeUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing barrier size.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_barrier_size(barrier_size_id, data)
if not result:
raise HTTPException(status_code=404, detail="Barrier size not found")
return result
@router.post("/product-applications", response_model=ProductApplicationOut, status_code=status.HTTP_201_CREATED)
def create_product_application(
data: ProductApplicationCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new product application.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_product_application(data)
@router.put("/product-applications/{application_id}", response_model=ProductApplicationOut)
def update_product_application(
application_id: int,
data: ProductApplicationUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing product application.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_product_application(application_id, data)
if not result:
raise HTTPException(status_code=404, detail="Product application not found")
return result
@router.post("/fobs", response_model=FOBOut, status_code=status.HTTP_201_CREATED)
def create_fob(
data: FOBCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new FOB term.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_fob(data)
@router.put("/fobs/{fob_id}", response_model=FOBOut)
def update_fob(
fob_id: int,
data: FOBUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing FOB term.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_fob(fob_id, data)
if not result:
raise HTTPException(status_code=404, detail="FOB not found")
return result
@router.post("/est-ship-dates", response_model=EstShipDateOut, status_code=status.HTTP_201_CREATED)
def create_est_ship_date(
data: EstShipDateCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new estimated ship date option.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_est_ship_date(data)
@router.put("/est-ship-dates/{est_ship_date_id}", response_model=EstShipDateOut)
def update_est_ship_date(
est_ship_date_id: int,
data: EstShipDateUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing estimated ship date option.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_est_ship_date(est_ship_date_id, data)
if not result:
raise HTTPException(status_code=404, detail="Estimated ship date not found")
return result
@router.post("/est-freights", response_model=EstFreightOut, status_code=status.HTTP_201_CREATED)
def create_est_freight(
data: EstFreightCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new estimated freight option.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_est_freight(data)
@router.put("/est-freights/{est_freight_id}", response_model=EstFreightOut)
def update_est_freight(
est_freight_id: int,
data: EstFreightUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing estimated freight option.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_est_freight(est_freight_id, data)
if not result:
raise HTTPException(status_code=404, detail="Estimated freight not found")
return result
@router.post("/status-info", response_model=StatusInfoOut, status_code=status.HTTP_201_CREATED)
def create_status_info(
data: StatusInfoCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new status info record.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_status_info(data)
@router.put("/status-info/{status_info_id}", response_model=StatusInfoOut)
def update_status_info(
status_info_id: int,
data: StatusInfoUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing status info record.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_status_info(status_info_id, data)
if not result:
raise HTTPException(status_code=404, detail="Status info not found")
return result
@router.post("/priorities", response_model=PriorityOut, status_code=status.HTTP_201_CREATED)
def create_priority(
data: PriorityCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Create a new priority.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
return service.create_priority(data)
@router.put("/priorities/{priority_id}", response_model=PriorityOut)
def update_priority(
priority_id: int,
data: PriorityUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""
Update an existing priority.
Requires authentication for production use.
"""
service = ReferenceDataService(db)
result = service.update_priority(priority_id, data)
if not result:
raise HTTPException(status_code=404, detail="Priority not found")
return result
@router.delete("/states/{state_id}", status_code=204)
def delete_state(
state_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a state"""
service = ReferenceDataService(db)
success = service.delete_state(state_id)
if not success:
raise HTTPException(status_code=404, detail="State not found")
return Response(status_code=204)
@router.delete("/countries/{country_id}", status_code=204)
def delete_country(
country_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a country"""
service = ReferenceDataService(db)
success = service.delete_country(country_id)
if not success:
raise HTTPException(status_code=404, detail="Country not found")
return Response(status_code=204)
@router.delete("/company-types/{company_type_id}", status_code=204)
def delete_company_type(
company_type_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a company type"""
service = ReferenceDataService(db)
success = service.delete_company_type(company_type_id)
if not success:
raise HTTPException(status_code=404, detail="Company type not found")
return Response(status_code=204)
@router.delete("/lead-sources/{lead_generated_from_id}", status_code=204)
def delete_lead_source(
lead_generated_from_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a lead source"""
service = ReferenceDataService(db)
success = service.delete_lead_source(lead_generated_from_id)
if not success:
raise HTTPException(status_code=404, detail="Lead source not found")
return Response(status_code=204)
@router.delete("/payment-terms/{payment_term_id}", status_code=204)
def delete_payment_term(
payment_term_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a payment term"""
service = ReferenceDataService(db)
success = service.delete_payment_term(payment_term_id)
if not success:
raise HTTPException(status_code=404, detail="Payment term not found")
return Response(status_code=204)
@router.delete("/purchase-prices/{purchase_price_id}", status_code=204)
def delete_purchase_price(
purchase_price_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a purchase price"""
service = ReferenceDataService(db)
success = service.delete_purchase_price(purchase_price_id)
if not success:
raise HTTPException(status_code=404, detail="Purchase price not found")
return Response(status_code=204)
@router.delete("/rental-prices/{rental_price_id}", status_code=204)
def delete_rental_price(
rental_price_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a rental price"""
service = ReferenceDataService(db)
success = service.delete_rental_price(rental_price_id)
if not success:
raise HTTPException(status_code=404, detail="Rental price not found")
return Response(status_code=204)
@router.delete("/barrier-sizes/{barrier_size_id}", status_code=204)
def delete_barrier_size(
barrier_size_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a barrier size"""
service = ReferenceDataService(db)
success = service.delete_barrier_size(barrier_size_id)
if not success:
raise HTTPException(status_code=404, detail="Barrier size not found")
return Response(status_code=204)
@router.delete("/product-applications/{application_id}", status_code=204)
def delete_product_application(
application_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a product application"""
service = ReferenceDataService(db)
success = service.delete_product_application(application_id)
if not success:
raise HTTPException(status_code=404, detail="Product application not found")
return Response(status_code=204)
@router.delete("/fobs/{fob_id}", status_code=204)
def delete_fob(
fob_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a FOB"""
service = ReferenceDataService(db)
success = service.delete_fob(fob_id)
if not success:
raise HTTPException(status_code=404, detail="FOB not found")
return Response(status_code=204)
@router.delete("/est-ship-dates/{est_ship_date_id}", status_code=204)
def delete_est_ship_date(
est_ship_date_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete an estimated ship date"""
service = ReferenceDataService(db)
success = service.delete_est_ship_date(est_ship_date_id)
if not success:
raise HTTPException(status_code=404, detail="Estimated ship date not found")
return Response(status_code=204)
@router.delete("/est-freights/{est_freight_id}", status_code=204)
def delete_est_freight(
est_freight_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete an estimated freight"""
service = ReferenceDataService(db)
success = service.delete_est_freight(est_freight_id)
if not success:
raise HTTPException(status_code=404, detail="Estimated freight not found")
return Response(status_code=204)
@router.delete("/status-info/{status_info_id}", status_code=204)
def delete_status_info(
status_info_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a status info"""
service = ReferenceDataService(db)
success = service.delete_status_info(status_info_id)
if not success:
raise HTTPException(status_code=404, detail="Status info not found")
return Response(status_code=204)
@router.delete("/priorities/{priority_id}", status_code=204)
def delete_priority(
priority_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a priority"""
service = ReferenceDataService(db)
success = service.delete_priority(priority_id)
if not success:
raise HTTPException(status_code=404, detail="Priority not found")
return Response(status_code=204)
@router.get("/phone-types", response_model=List[PhoneTypeOut])
def get_phone_types(
active_only: bool = Query(False, description="Return only active records"),
db: Session = Depends(get_db),
current_user: Optional[CurrentUser] = Depends(get_current_user_optional)
):
"""Get all phone types"""
service = ReferenceDataService(db)
return service.get_phone_types(active_only)
@router.post("/phone-types", response_model=PhoneTypeOut, status_code=201)
def create_phone_type(
data: PhoneTypeCreateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Create a new phone type"""
service = ReferenceDataService(db)
return service.create_phone_type(data)
@router.put("/phone-types/{phone_type_id}", response_model=PhoneTypeOut)
def update_phone_type(
phone_type_id: int,
data: PhoneTypeUpdateIn,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Update an existing phone type"""
service = ReferenceDataService(db)
result = service.update_phone_type(phone_type_id, data)
if not result:
raise HTTPException(status_code=404, detail="Phone type not found")
return result
@router.delete("/phone-types/{phone_type_id}", status_code=204)
def delete_phone_type(
phone_type_id: int,
db: Session = Depends(get_db),
current_user: CurrentUser = Depends(get_current_user_optional)
):
"""Delete a phone type"""
service = ReferenceDataService(db)
success = service.delete_phone_type(phone_type_id)
if not success:
raise HTTPException(status_code=404, detail="Phone type not found")
return Response(status_code=204)