Okay, let's switch to Python using FastAPI. This framework is well-suited for building APIs efficiently. We'll also incorporate the requirement to disable SSL verification for requests when needed, managed via an environment variable. This setup includes: * A FastAPI application. * Configuration loading from a `.env` file using `python-dotenv`. * A reusable HTTP client utility that handles base URLs, headers, and the `verify=False` setting. * Structured routers, controllers (logic handlers), and services (external API interaction). * Placeholder logic for interacting with Collateral360 and your internal LOS. fastapi uvicorn[standard] requests python-dotenv pydantic # Server Configuration APP_HOST=0.0.0.0 APP_PORT=8000 # Collateral360 API Configuration COLLATERAL360_API_BASE_URL=https://api.lightboxre.com/collateral360/v1 # Replace with actual base URL if different COLLATERAL360_API_KEY=YOUR_LIGHTBOX_AUTHENTICATION_KEY_HERE # Internal LOS Configuration (Example) INTERNAL_LOS_API_BASE_URL=http://your-internal-los-api.local/api # Replace with actual internal LOS API URL INTERNAL_LOS_API_KEY=YOUR_INTERNAL_LOS_API_KEY_HERE # If needed # SSL Verification Control (set to False for non-prod/proxy issues) REQUESTS_VERIFY_SSL=False # This file makes 'src' a Python package import os from dotenv import load_dotenv from pydantic_settings import BaseSettings # Load .env file variables into environment load_dotenv() class Settings(BaseSettings): APP_HOST: str = "0.0.0.0" APP_PORT: int = 8000 COLLATERAL360_API_BASE_URL: str COLLATERAL360_API_KEY: str INTERNAL_LOS_API_BASE_URL: str INTERNAL_LOS_API_KEY: str | None = None # Optional API key # Read SSL verification flag, default to True if not set or invalid REQUESTS_VERIFY_SSL_STR: str = os.getenv("REQUESTS_VERIFY_SSL", "True") @property def REQUESTS_VERIFY_SSL(self) -> bool: return self.REQUESTS_VERIFY_SSL_STR.lower() in ("true", "1", "t", "yes") class Config: # This tells pydantic-settings to load from environment variables # It will automatically pick up variables defined above env_file = '.env' env_file_encoding = 'utf-8' extra = 'ignore' # Ignore extra fields in .env settings = Settings() # print(f"SSL Verification Enabled: {settings.REQUESTS_VERIFY_SSL}") # For debugging # This file makes 'utils' a Python package import requests from requests import Session from src.config import settings import logging logger = logging.getLogger(__name__) def get_requests_session() -> Session: """Creates a reusable requests session.""" session = requests.Session() session.verify = settings.REQUESTS_VERIFY_SSL if not settings.REQUESTS_VERIFY_SSL: # Suppress only the single InsecureRequestWarning from urllib3 needed import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) logger.warning("SSL verification disabled for requests.") return session # --- Client for Collateral360 --- def get_collateral360_client() -> Session: session = get_requests_session() session.base_url = settings.COLLATERAL360_API_BASE_URL # type: ignore [attr-defined] # Add base_url attribute session.headers.update({ 'Content-Type': 'application/json', # ** CONSULT COLLATERAL360 API DOCUMENTATION FOR AUTHENTICATION ** # Adjust header based on actual C360 requirement (e.g., 'Authorization': f'Bearer {settings.COLLATERAL360_API_KEY}') 'X-Api-Key': settings.COLLATERAL360_API_KEY }) # Add hooks or adapters if needed for advanced request modification return session # --- Client for Internal LOS (Example) --- def get_internal_los_client() -> Session: session = get_requests_session() session.base_url = settings.INTERNAL_LOS_API_BASE_URL # type: ignore [attr-defined] # Add base_url attribute session.headers.update({ 'Content-Type': 'application/json', }) # Add Authorization header if needed if settings.INTERNAL_LOS_API_KEY: session.headers.update({'Authorization': f'Bearer {settings.INTERNAL_LOS_API_KEY}'}) # Example Bearer token return session # --- Helper to make requests with base URL --- def make_request(session: Session, method: str, endpoint: str, **kwargs): """Makes a request using the session's base URL.""" url = f"{session.base_url}{endpoint}" # type: ignore [attr-defined] logger.debug(f"Making request: {method.upper()} {url}") try: response = session.request(method, url, **kwargs) response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) return response.json() except requests.exceptions.RequestException as e: logger.error(f"HTTP Request failed: {e}") # Attempt to get more detail from response if available error_detail = None if e.response is not None: try: error_detail = e.response.json() except ValueError: # Includes JSONDecodeError error_detail = e.response.text logger.error(f"Error details: {error_detail}") # Re-raise a more specific or generic error as needed for your app raise e # Or wrap in a custom exception # This file makes 'services' a Python package from src.utils.http_client import get_collateral360_client, make_request import logging logger = logging.getLogger(__name__) client = get_collateral360_client() async def fetch_loan_file(loan_id: str) -> dict: """Fetches loan file information from Collateral360.""" endpoint = f"/loans/{loan_id}" # Adjust endpoint as per C360 docs logger.info(f"Calling Collateral360 API: GET {endpoint}") try: # Using the helper function for consistency response_data = make_request(client, "GET", endpoint) return response_data except Exception as e: logger.error(f"Collateral360 API Error fetching loan {loan_id}: {e}") # Depending on desired error handling, you might return None, an error dict, or raise a custom exception raise Exception(f"Failed to fetch loan file from Collateral360: {e}") async def submit_service_request(request_data: dict) -> dict: """Submits a service request to Collateral360.""" endpoint = "/services" # Adjust endpoint as per C360 docs logger.info(f"Calling Collateral360 API: POST {endpoint}") try: response_data = make_request(client, "POST", endpoint, json=request_data) return response_data except Exception as e: logger.error(f"Collateral360 API Error submitting service request: {e}") raise Exception(f"Failed to submit service request to Collateral360: {e}") # --- Add more functions corresponding to C360 API features --- # async def fetch_environmental_risk_codes(): ... # async def add_collateral(loan_id: str, collateral_data: dict): ... # async def add_services(loan_id: str, service_data: dict): ... # async def check_service_status(service_id: str): ... # async def fetch_downloadable_files(loan_id: str): ... from src.utils.http_client import get_internal_los_client, make_request import logging logger = logging.getLogger(__name__) client = get_internal_los_client() # --- Placeholder Service for Internal LOS Interaction --- async def get_loan_details(loan_id: str) -> dict: """(Placeholder) Fetches loan details from the Internal LOS.""" logger.info(f"(Placeholder) Fetching details for loan {loan_id} from Internal LOS") # Replace with actual API call # endpoint = f"/loans/{loan_id}" # try: # return make_request(client, "GET", endpoint) # except Exception as e: # logger.error(f"Internal LOS API Error fetching loan {loan_id}: {e}") # raise Exception(f"Failed to fetch loan details from Internal LOS: {e}") # Return mock data for now return { "internalId": loan_id, "borrowerName": "Placeholder Borrower (Python)", "status": "Pending Orchestration" } async def update_loan_status(loan_id: str, status: str) -> None: """(Placeholder) Updates loan status in the Internal LOS.""" logger.info(f'(Placeholder) Updating status for loan {loan_id} to "{status}" in Internal LOS') # Replace with actual API call # endpoint = f"/loans/{loan_id}/status" # try: # make_request(client, "PATCH", endpoint, json={"status": status}) # Or PUT/POST # logger.info(f"Internal LOS: Loan {loan_id} status updated successfully.") # except Exception as e: # logger.error(f"Internal LOS API Error updating status for loan {loan_id}: {e}") # raise Exception(f"Failed to update loan status in Internal LOS: {e}") return None # Indicate success async def update_loan_with_c360_result(loan_id: str, c360_response: dict) -> None: """(Placeholder) Updates loan with C360 results in Internal LOS.""" logger.info(f"(Placeholder) Updating loan {loan_id} with C360 results in Internal LOS") # Replace with actual API call # endpoint = f"/loans/{loan_id}/c360data" # try: # make_request(client, "PUT", endpoint, json={"c360Data": c360_response}) # Or POST/PATCH # logger.info(f"Internal LOS: Loan {loan_id} updated with C360 data successfully.") # except Exception as e: # logger.error(f"Internal LOS API Error updating loan {loan_id} with C360 data: {e}") # raise Exception(f"Failed to update loan with C360 data in Internal LOS: {e}") return None # Indicate success # --- Add more functions as needed --- # This file makes 'controllers' a Python package from fastapi import HTTPException from src.services import collateral360_service, internal_los_service import logging from pydantic import BaseModel, Field # For request body validation logger = logging.getLogger(__name__) # --- Request Body Models (using Pydantic) --- class ServiceRequestPayload(BaseModel): # Define expected fields for creating a service request # Example fields - adjust based on actual C360 requirements loanId: str = Field(..., description="The unique identifier for the loan") serviceType: str = Field(..., description="Type of service being requested") # Add other necessary fields... # requester_email: str | None = None # --- Controller Functions --- async def get_loan_file_orchestration(loan_id: str): """Orchestrates fetching loan file info from C360 and internal LOS.""" logger.info(f"Orchestrating request for loan file: {loan_id}") try: # 1. Potentially fetch data from internal LOS first (if needed) # internal_data = await internal_los_service.get_loan_details(loan_id) # logger.info(f'Fetched data from internal LOS: {internal_data}') # 2. Fetch data from Collateral360 API collateral_data = await collateral360_service.fetch_loan_file(loan_id) logger.info(f'Fetched data from Collateral360: {collateral_data}') # 3. Combine/transform data as needed response_data = { "source": "Orchestration API (Python)", "collateral360Info": collateral_data, # "internalLosInfo": internal_data # Include if fetched } return response_data except Exception as e: logger.error(f"Error in get_loan_file_orchestration for {loan_id}: {e}") # Re-raise as HTTPException for FastAPI to handle raise HTTPException(status_code=500, detail=f"Failed to orchestrate loan file retrieval: {e}") async def create_service_request_orchestration(payload: ServiceRequestPayload): """Orchestrates creating a service request via Collateral360.""" logger.info(f"Orchestrating creation of a service request for loan: {payload.loanId}") try: # 1. Convert Pydantic model to dict for the service layer service_request_data = payload.model_dump() # 2. Potentially interact with internal LOS before C360 (e.g., update status) # await internal_los_service.update_loan_status(payload.loanId, 'PENDING_C360_SERVICE') # 3. Submit request to Collateral360 API collateral_response = await collateral360_service.submit_service_request(service_request_data) logger.info(f'Collateral360 response: {collateral_response}') # 4. Potentially update internal LOS with C360 response details # await internal_los_service.update_loan_with_c360_result(payload.loanId, collateral_response) return { "message": "Service request submitted successfully via orchestration.", "collateral360Response": collateral_response } except Exception as e: logger.error(f"Error in create_service_request_orchestration: {e}") # Re-raise as HTTPException raise HTTPException(status_code=500, detail=f"Failed to orchestrate service request creation: {e}") # --- Add more controller functions for other routes --- # This file makes 'routers' a Python package from fastapi import APIRouter, Depends, Body # Added Body for explicit payload marking from src.controllers import collateral_controller from src.controllers.collateral_controller import ServiceRequestPayload # Import the Pydantic model router = APIRouter( prefix="/api/collateral", # Base path for this router tags=["Collateral Orchestration"], # Tag for OpenAPI docs ) # Example Route: Get Loan File Info @router.get("/loans/{loan_id}", response_model=dict) # Define response model if known async def get_loan_file(loan_id: str): """ Retrieves orchestrated loan file information, potentially combining data from Collateral360 and the internal LOS. """ return await collateral_controller.get_loan_file_orchestration(loan_id) # Example Route: Create a Service Request @router.post("/services", status_code=201, response_model=dict) # Set default success status code async def create_service_request( payload: ServiceRequestPayload = Body(...) # Use the Pydantic model for validation ): """ Submits a new service request through the orchestration layer, primarily interacting with the Collateral360 API. """ return await collateral_controller.create_service_request_orchestration(payload) # --- Add more routes based on Collateral360 API features --- # e.g., @router.get("/codes/environmental") # async def get_environmental_codes(): ... # e.g., @router.post("/loans/{loan_id}/collateral") # async def add_collateral(loan_id: str, collateral_data: dict = Body(...)): ... # e.g., @router.post("/loans/{loan_id}/services") # async def add_services_to_loan(loan_id: str, service_data: dict = Body(...)): ... # e.g., @router.get("/services/{service_id}/status") # async def check_service_status(service_id: str): ... # e.g., @router.get("/loans/{loan_id}/files") # async def get_downloadable_files(loan_id: str): ... import uvicorn from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from fastapi.exceptions import RequestValidationError from src.routers import collateral from src.config import settings import logging import sys # --- Logging Configuration --- # Basic configuration, adjust level and format as needed logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', stream=sys.stdout, # Log to stdout ) logger = logging.getLogger(__name__) # --- FastAPI App Initialization --- app = FastAPI( title="Collateral360 Orchestration API", description="Orchestrates interactions between a 3rd party app and Collateral360 / Internal LOS.", version="1.0.0", ) # --- Include Routers --- app.include_router(collateral.router) # --- Basic Root Route --- @app.get("/") async def root(): return {"message": "Collateral360 Orchestration API (Python/FastAPI) is running!"} # --- Exception Handlers (Optional but Recommended) --- @app.exception_handler(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError): logger.error(f"Validation error: {exc.errors()}") return JSONResponse( status_code=422, content={"detail": exc.errors(), "body": exc.body}, ) @app.exception_handler(Exception) async def generic_exception_handler(request: Request, exc: Exception): logger.exception(f"Unhandled exception during request to {request.url}: {exc}") # Logs traceback return JSONResponse( status_code=500, content={"detail": "An internal server error occurred."}, ) # --- Run Server (for direct execution, e.g., python main.py) --- if __name__ == "__main__": logger.info(f"Starting server on {settings.APP_HOST}:{settings.APP_PORT}") logger.info(f"SSL Verification for outgoing requests: {settings.REQUESTS_VERIFY_SSL}") uvicorn.run( "main:app", host=settings.APP_HOST, port=settings.APP_PORT, reload=True # Enable auto-reload for development ) **To Run This Project:** 1. **Install Dependencies:** bash pip install -r requirements.txt 2. **Configure Environment:** * Create or update the `.env` file in the project root. * Fill in your `COLLATERAL360_API_BASE_URL` and `COLLATERAL360_API_KEY`. * Fill in your `INTERNAL_LOS_API_BASE_URL` and `INTERNAL_LOS_API_KEY` (if applicable). * Set `REQUESTS_VERIFY_SSL=False` if you need to bypass SSL certificate checks (e.g., due to a proxy or self-signed certs in non-prod). Otherwise, set it to `True` or remove the line (it defaults to True if missing). * **Crucially:** Check the Collateral360 API documentation for the exact authentication method (e.g., Bearer token, specific header name) and update the `X-Api-Key` header in `src/utils/http_client.py` accordingly. 3. **Implement Logic:** * Update the API endpoints and logic in `src/services/collateral360_service.py` based on the actual Collateral360 documentation. * Replace placeholder functions in `src/services/internal_los_service.py` with real calls to your internal system. * Refine orchestration logic in `src/controllers/collateral_controller.py`. * Define accurate request/response models using Pydantic in the controller or a separate `models.py` file for better validation and documentation. * Add more routes, controllers, and service functions as needed. 4. **Run the Server:** ```bash uvicorn main:app --host 0.0.0.0 --port 8000 --reload ``` * `--reload` automatically restarts the server when code changes (useful for development). * Access the API at `http://localhost:8000` (or the host/port you configure). * Interactive API documentation (Swagger UI) will be available at `http://localhost:8000/docs`. * Alternative documentation (ReDoc) at `http://localhost:8000/redoc`. This FastAPI structure provides a robust and scalable foundation for your orchestration API, addressing your Python preference and the SSL verification requirement.