Spaces:
Paused
Paused
| 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"]) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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"} | |
| 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) | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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 | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |
| 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 | |
| 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) |