AlexeerCT
d69dd23bcd
- Updated validation logic to restrict 'IMD' document type usage unless the invoice type is 'DEF'. - Refactored value assignment in the main processing flow to handle 'DEF' and 'MEX' invoice types with specific IVA calculations. - Added logging for invoice processing to improve traceability and debugging. These changes improve the accuracy of invoice validations and processing for specific document types.
119 lines
3.8 KiB
Python
119 lines
3.8 KiB
Python
"""
|
|
SITAR API Base Service
|
|
|
|
Base service class for SITAR API authentication and HTTP requests.
|
|
All specific resource services inherit from this.
|
|
"""
|
|
|
|
import os
|
|
from typing import Optional, Dict, Any
|
|
from datetime import datetime, timedelta
|
|
import httpx
|
|
|
|
|
|
class SitarAPIBaseService:
|
|
"""Base service for SITAR API integration with authentication"""
|
|
|
|
_token: Optional[str] = None
|
|
_token_expires: Optional[datetime] = None
|
|
|
|
def __init__(self):
|
|
"""Initialize base service with API credentials"""
|
|
self.base_url = os.getenv("SITAR_API_URL")
|
|
self.username = os.getenv("SITAR_API_USER")
|
|
self.password = os.getenv("SITAR_API_PASSWORD")
|
|
self.timeout = 30.0
|
|
|
|
if not all([self.base_url, self.username, self.password]):
|
|
raise ValueError(
|
|
"Missing SITAR API configuration. "
|
|
"Set SITAR_API_URL, SITAR_API_USER, and SITAR_API_PASSWORD environment variables."
|
|
)
|
|
|
|
async def _get_token(self) -> str:
|
|
"""
|
|
Get authentication token, refreshing if necessary
|
|
|
|
Returns:
|
|
str: Bearer token for API authentication
|
|
|
|
Raises:
|
|
httpx.HTTPError: If authentication fails
|
|
"""
|
|
# Return cached token if still valid
|
|
if self._token and self._token_expires and datetime.now() < self._token_expires:
|
|
return self._token
|
|
|
|
# Authenticate and get new token
|
|
login_url = f"{self.base_url}/fractions/api/v1/auth/login"
|
|
payload = {"username": self.username, "password": self.password}
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.post(login_url, json=payload)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
self._token = data.get("access_token") or data.get("token")
|
|
|
|
if not self._token:
|
|
raise ValueError("No token received from SITAR API")
|
|
|
|
# Set token expiration (assume 1 hour if not specified)
|
|
self._token_expires = datetime.now() + timedelta(hours=1)
|
|
|
|
return self._token
|
|
|
|
async def _make_request(
|
|
self,
|
|
method: str,
|
|
endpoint: str,
|
|
params: Optional[Dict[str, Any]] = None,
|
|
json_data: Optional[Dict[str, Any]] = None,
|
|
) -> Any:
|
|
"""
|
|
Make authenticated request to SITAR API
|
|
|
|
Args:
|
|
method: HTTP method (GET, POST, etc.)
|
|
endpoint: API endpoint path
|
|
params: Query parameters
|
|
json_data: JSON body data
|
|
|
|
Returns:
|
|
JSON response data
|
|
|
|
Raises:
|
|
httpx.HTTPError: If request fails
|
|
"""
|
|
token = await self._get_token()
|
|
# Ensure no double slash between fractures and endpoint
|
|
endpoint = endpoint.lstrip("/")
|
|
url = f"{self.base_url}/fractions/{endpoint}"
|
|
headers = {
|
|
"Authorization": f"Bearer {token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
|
response = await client.request(
|
|
method=method,
|
|
url=url,
|
|
params=params,
|
|
json=json_data,
|
|
headers=headers,
|
|
)
|
|
response.raise_for_status()
|
|
|
|
# DEBUG LOGGING for SITAR inspection
|
|
if "fracciones" in url:
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
try:
|
|
data = response.json()
|
|
return data
|
|
except Exception:
|
|
logger.error(f"Error parsing SITAR API Response Body: {response.text}")
|
|
pass
|
|
|
|
return response.json()
|