feat: add historical tariff fractions functionality
- Implemented new endpoints for managing historical tariff fractions, including listing and fetching rates based on invoice details. - Created DTOs for historical tariff fractions to standardize data transfer. - Added service layer to handle business logic for historical tariff fractions. - Updated frontend components to integrate historical tariff fractions, including fetching rates based on user input. - Enhanced existing data models and routes to accommodate new functionality.
This commit is contained in:
@@ -0,0 +1,29 @@
|
||||
"""
|
||||
DTOs for historical tariff fractions.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
from typing import Optional
|
||||
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
|
||||
class HistoricalTariffFractionResponseDTO(BaseModel):
|
||||
id: int
|
||||
historical_fraction: Optional[str] = None
|
||||
unit_of_measure_code: Optional[str] = None
|
||||
country: Optional[str] = None
|
||||
fraction_type: Optional[str] = None
|
||||
sector: Optional[str] = None
|
||||
import_tax_rate: Optional[Decimal] = None
|
||||
export_tax_rate: Optional[Decimal] = None
|
||||
publication_date: Optional[datetime] = None
|
||||
is_immex: Optional[bool] = None
|
||||
normal_temporality: Optional[bool] = None
|
||||
services_temporality: Optional[bool] = None
|
||||
certified_temporality: Optional[bool] = None
|
||||
by_log: Optional[bool] = None
|
||||
end_date: Optional[datetime] = None
|
||||
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
@@ -17,6 +17,7 @@ class HistoricalTariffFraction(Base):
|
||||
|
||||
id: Mapped[int] = mapped_column(Integer, primary_key=True, nullable=False)
|
||||
historical_fraction: Mapped[Optional[str]] = mapped_column(String(8), nullable=True)
|
||||
nico: Mapped[Optional[str]] = mapped_column(String(2), nullable=True)
|
||||
unit_of_measure_code: Mapped[Optional[str]] = mapped_column(ForeignKey("a76.unit_of_measure_customs.code"), nullable=True)
|
||||
country: Mapped[Optional[str]] = mapped_column(ForeignKey("public.countries.m3_key"), nullable=True)
|
||||
fraction_type: Mapped[Optional[str]] = mapped_column(String(7), nullable=True)
|
||||
|
||||
@@ -0,0 +1,128 @@
|
||||
"""
|
||||
Endpoints for historical tariff fractions.
|
||||
"""
|
||||
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from core.database import get_core_db
|
||||
from core.security import get_current_user
|
||||
|
||||
from .dto import HistoricalTariffFractionResponseDTO
|
||||
from .service import HistoricalTariffFractionService
|
||||
|
||||
router = APIRouter(
|
||||
prefix="/historical-tariff-fractions",
|
||||
tags=["a76 / general catalogs / historical tariff fractions"],
|
||||
)
|
||||
|
||||
|
||||
@router.get(
|
||||
"/",
|
||||
response_model=Dict[str, Any],
|
||||
summary="List Historical Tariff Fractions",
|
||||
description=(
|
||||
"Get paginated list of historical tariff fractions with optional filters"
|
||||
),
|
||||
)
|
||||
async def list_historical_tariff_fractions(
|
||||
page: int = Query(1, ge=1, description="Page number"),
|
||||
page_size: int = Query(50, ge=1, le=10000, description="Page size"),
|
||||
historical_fraction: Optional[str] = Query(
|
||||
None, description="Filter by historical fraction"
|
||||
),
|
||||
nico: Optional[str] = Query(None, description="Filter by NICO"),
|
||||
country: Optional[str] = Query(None, description="Filter by country"),
|
||||
publication_date: Optional[str] = Query(
|
||||
None, description="Filter by publication date (YYYY-MM-DD)"
|
||||
),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
):
|
||||
skip = (page - 1) * page_size
|
||||
filters: Dict[str, Any] = {}
|
||||
|
||||
if historical_fraction:
|
||||
filters["historical_fraction"] = historical_fraction
|
||||
if nico:
|
||||
filters["nico"] = nico
|
||||
if country:
|
||||
filters["country"] = country
|
||||
if publication_date:
|
||||
filters["publication_date"] = publication_date
|
||||
|
||||
items, total = HistoricalTariffFractionService.get_all(db, skip, page_size, filters)
|
||||
|
||||
return {
|
||||
"items": [
|
||||
HistoricalTariffFractionResponseDTO.model_validate(item) for item in items
|
||||
],
|
||||
"total": total,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
"pages": (total + page_size - 1) // page_size,
|
||||
}
|
||||
|
||||
|
||||
@router.get(
|
||||
"/rate",
|
||||
response_model=Dict[str, Any],
|
||||
summary="Get Historical Tariff Rate for Invoice",
|
||||
description=(
|
||||
"Get the import tax rate for a historical tariff fraction based on invoice date. "
|
||||
"Replicates Clarion ASIGNA_FRACCION_HISTORICA logic: returns the most recent "
|
||||
"fraction published before or on the invoice date."
|
||||
),
|
||||
)
|
||||
async def get_historical_tariff_rate(
|
||||
historical_fraction: str = Query(
|
||||
...,
|
||||
min_length=8,
|
||||
max_length=8,
|
||||
description="Historical fraction (8 characters)",
|
||||
),
|
||||
nico: str = Query(
|
||||
..., min_length=2, max_length=2, description="nico code (2 characters)"
|
||||
),
|
||||
fraction_type: str = Query(..., description="Fraction type (GENERAL, TLCS, etc.)"),
|
||||
invoice_date: str = Query(..., description="Invoice date in YYYY-MM-DD format"),
|
||||
sector: Optional[str] = Query(None, description="Sector (optional, for PROSEC)"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
):
|
||||
item = HistoricalTariffFractionService.get_rate_for_invoice(
|
||||
db, historical_fraction, nico, fraction_type, invoice_date, sector
|
||||
)
|
||||
if not item:
|
||||
return {"rate": None, "found": False}
|
||||
|
||||
return {
|
||||
"rate": float(item.import_tax_rate) if item.import_tax_rate else 0.0,
|
||||
"found": True,
|
||||
"publication_date": (
|
||||
item.publication_date.isoformat() if item.publication_date else None
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
@router.get(
|
||||
"/{historical_tariff_fraction_id}",
|
||||
response_model=HistoricalTariffFractionResponseDTO,
|
||||
summary="Get Historical Tariff Fraction by ID",
|
||||
description="Get a specific historical tariff fraction by ID",
|
||||
)
|
||||
async def get_historical_tariff_fraction(
|
||||
historical_tariff_fraction_id: int,
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
):
|
||||
item = HistoricalTariffFractionService.get_by_id(db, historical_tariff_fraction_id)
|
||||
if not item:
|
||||
from fastapi import HTTPException
|
||||
|
||||
raise HTTPException(
|
||||
status_code=404, detail="Historical tariff fraction not found"
|
||||
)
|
||||
return HistoricalTariffFractionResponseDTO.model_validate(item)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,111 @@
|
||||
"""
|
||||
Service for historical tariff fractions (global catalog).
|
||||
"""
|
||||
|
||||
from datetime import datetime, time
|
||||
from typing import List, Optional, Tuple, Dict, Any
|
||||
|
||||
from sqlalchemy import cast, Date
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from .models import HistoricalTariffFraction
|
||||
|
||||
|
||||
class HistoricalTariffFractionService:
|
||||
@staticmethod
|
||||
def get_all(
|
||||
db: Session,
|
||||
skip: int = 0,
|
||||
limit: int = 100,
|
||||
filters: Optional[Dict[str, Any]] = None,
|
||||
) -> Tuple[List[HistoricalTariffFraction], int]:
|
||||
query = db.query(HistoricalTariffFraction)
|
||||
|
||||
if filters:
|
||||
if filters.get("historical_fraction"):
|
||||
term = filters["historical_fraction"]
|
||||
query = query.filter(
|
||||
HistoricalTariffFraction.historical_fraction.ilike(f"%{term}%")
|
||||
)
|
||||
if filters.get("nico"):
|
||||
term = filters["nico"]
|
||||
query = query.filter(
|
||||
HistoricalTariffFraction.historical_fraction.ilike(f"%{term}%")
|
||||
)
|
||||
if filters.get("country"):
|
||||
term = filters["country"]
|
||||
query = query.filter(
|
||||
HistoricalTariffFraction.country.ilike(f"%{term}%")
|
||||
)
|
||||
if filters.get("publication_date"):
|
||||
try:
|
||||
date_str = filters["publication_date"]
|
||||
date_val = datetime.strptime(date_str, "%Y-%m-%d").date()
|
||||
start_date = datetime.combine(date_val, time.min)
|
||||
end_date = datetime.combine(date_val, time.max)
|
||||
query = query.filter(
|
||||
HistoricalTariffFraction.publication_date >= start_date,
|
||||
HistoricalTariffFraction.publication_date <= end_date,
|
||||
)
|
||||
except (ValueError, TypeError):
|
||||
query = query.filter(
|
||||
cast(HistoricalTariffFraction.publication_date, Date)
|
||||
== filters["publication_date"]
|
||||
)
|
||||
|
||||
total = query.count()
|
||||
items = (
|
||||
query.order_by(HistoricalTariffFraction.publication_date.desc())
|
||||
.offset(skip)
|
||||
.limit(limit)
|
||||
.all()
|
||||
)
|
||||
|
||||
return items, total
|
||||
|
||||
@staticmethod
|
||||
def get_by_id(
|
||||
db: Session,
|
||||
historical_tariff_fraction_id: int,
|
||||
) -> Optional[HistoricalTariffFraction]:
|
||||
return (
|
||||
db.query(HistoricalTariffFraction)
|
||||
.filter(HistoricalTariffFraction.id == historical_tariff_fraction_id)
|
||||
.first()
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def get_rate_for_invoice(
|
||||
db: Session,
|
||||
historical_fraction: str,
|
||||
nico: str,
|
||||
fraction_type: str,
|
||||
invoice_date: str,
|
||||
sector: Optional[str] = None,
|
||||
) -> Optional[HistoricalTariffFraction]:
|
||||
"""
|
||||
Replicates Clarion ASIGNA_FRACCION_HISTORICA logic:
|
||||
SELECT TOP 1 WHERE FraccionHistorica = {fraction_8chars}
|
||||
AND Pais = {nico_2chars}
|
||||
AND FechaPublicacion <= {invoice_date}
|
||||
ORDER BY FechaPublicacion ASC
|
||||
"""
|
||||
try:
|
||||
date_val = datetime.strptime(invoice_date, "%Y-%m-%d").date()
|
||||
invoice_datetime = datetime.combine(date_val, time.max)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
print(f"Searching for historical tariff fraction with historical_fraction={historical_fraction}, nico={nico}, fraction_type={fraction_type}, invoice_date={invoice_datetime}, sector={sector}")
|
||||
|
||||
query = db.query(HistoricalTariffFraction).filter(
|
||||
HistoricalTariffFraction.historical_fraction == historical_fraction,
|
||||
HistoricalTariffFraction.nico == nico,
|
||||
HistoricalTariffFraction.fraction_type == fraction_type,
|
||||
HistoricalTariffFraction.publication_date <= invoice_datetime,
|
||||
)
|
||||
|
||||
if sector:
|
||||
query = query.filter(HistoricalTariffFraction.sector == sector)
|
||||
|
||||
return query.order_by(HistoricalTariffFraction.publication_date.asc()).first()
|
||||
@@ -6,6 +6,9 @@ from .packages.routes import router as package_router
|
||||
from .ports.routes import router as ports_router
|
||||
from .fractions.tariff_fractions.routes import router as tariff_fractions_router
|
||||
from .fractions.us_tariff_fractions.routes import router as us_tariff_fractions_router
|
||||
from .fractions.historical_tariff_fractions.routes import (
|
||||
router as historical_tariff_fractions_router,
|
||||
)
|
||||
from .depreciation_catalog.routes import router as depreciation_catalog_router
|
||||
from .fda_catalog.routes import router as fda_catalog_router
|
||||
from .seal.routes import router as seal_router
|
||||
@@ -26,14 +29,15 @@ from .electronic_notices.routes import router as electronic_notices_router
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
router.include_router(company_router, tags=["a76 / company"])
|
||||
router.include_router(company_router, tags=["a76 / company"])
|
||||
router.include_router(package_router)
|
||||
router.include_router(ports_router)
|
||||
router.include_router(tariff_fractions_router)
|
||||
router.include_router(us_tariff_fractions_router)
|
||||
router.include_router(historical_tariff_fractions_router)
|
||||
router.include_router(depreciation_catalog_router)
|
||||
router.include_router(fda_catalog_router)
|
||||
router.include_router(seal_router, tags=["a76 / seal"])
|
||||
router.include_router(seal_router, tags=["a76 / seal"])
|
||||
router.include_router(units_of_measure_router)
|
||||
router.include_router(identifiers_router)
|
||||
router.include_router(exchange_rate_router, tags=["a76 / exchange_rate"])
|
||||
@@ -49,4 +53,4 @@ router.include_router(signatures_router)
|
||||
router.include_router(error_catalogs_router)
|
||||
router.include_router(doda_router)
|
||||
router.include_router(prevalidators_router)
|
||||
router.include_router(electronic_notices_router)
|
||||
router.include_router(electronic_notices_router)
|
||||
|
||||
Reference in New Issue
Block a user