--- updatedAt: 2025-01-06T18:15:35.000Z agentTools: projectIndex: https://developer.truckstop.com/llms.txt --- # Manage onboarded carrier documents ```python DocumentAPI.py import requests import xml.etree.ElementTree as ET from RMISLibrary import RMISCredentials, Document from TruckStopDBHelper import * def GetDocument(rmisCreds: RMISCredentials, document: Document): # Open a connection to the database and check if this document already exists dateBaseObject = GetDBConnection("RMISCarriers") documentCheck = dateBaseObject.sqlCursor.execute(f'SELECT * FROM Documents where ClientID=\'{rmisCreds.clientID}\' AND DocumentID = \'{document.documentID}\'') documentCheck = dateBaseObject.sqlCursor.fetchall() if (len(documentCheck) == 0): # Document does not exist in the database # Base URL for the Document API base_url = "https://api.rmissecure.com/_c/std/api/DocumentAPI.aspx" # Parameters for the API request params = { "clientID": rmisCreds.clientID, # Client ID for authentication "pwd": rmisCreds.clientPassword, # Client password for authentication "documentType": document.documentType, # Type of document being requested "documentID": document.documentID, # Unique identifier for the document "insdID": document.CarrierRMISID, # Insured ID (carrier ID) related to the document "version": "1" # Version of the API being called } # Headers for the API request headers = {"accept": "application/xml"} # Indicate the response format expected # Make the API request with the specified parameters response = requests.get(base_url, headers=headers, params=params) # Check if the response was successful (HTTP status code 200) if response.status_code == 200: newDocument = __ProcessResponse(response) # Process the successful response newDocument.CarrierRMISID = document.CarrierRMISID newDocument.clientID = rmisCreds.clientID InsertIntoTable(table="Documents", dataBaseObject=dateBaseObject, object=newDocument) else: # Log the error details if the request was not successful print(f"Failed to retrieve document: {response.status_code} - {response.text}") return False # Close the database connection dateBaseObject.sqlConn.close() def __ProcessResponse(response: requests.Response): # Parse the XML response content to extract relevant fields root = ET.fromstring(response.content) try: # Extract relevant fields from the XML document_data = root.find('.//DocumentData').text # Base64-encoded document data document_file_type = root.find('.//DocumentFileType').text # Type of the document (e.g., PDF, JPG) document_title = root.find('.//DocumentTitle').text # Title of the document document_size_in_bytes = root.find('.//DocumentFileSizeInBytes').text # Size of the document document_type = root.find('.//DocumentType').text # Type of document (client agreement/etc) document_date = root.find('.//DocumentDate').text # Date the document was uploaded document_id = root.find('.//DocumentID').text # Specific ID for this document document = Document(documentType=document_type, documentID=document_id, carrierID=None) document.documentData = document_data document.documentFileType = document_file_type document.documentTitle = document_title document.documentSizeInBytes = document_size_in_bytes document.documentDate = document_date return document except Exception as e: # Handle exceptions and extract any errors from the response errors = root.find('.//Errors') # Check for errors in the response for error in errors: print(error.text) # Log each error message return False ``` ```python RMISLibrary.py class RMISCredentials: def __init__(self, clientID, password): self.clientID = clientID # Client ID for RMIS API self.clientPassword = password # Client password for RMIS API class Carrier: def __init__(self, rmisId = None, clientId=None, carrierName=None, carrierMC=None, carrierDOT=None, lastUpdated=None): self.CarrierRMISID=rmisId self.clientID = clientId self.carrierName = carrierName self.carrierMC = carrierMC self.carrierDOT = carrierDOT self.lastUpdated = lastUpdated #Class to represent a document class Document: def __init__(self, documentType, documentID, carrierID): self.documentType = documentType # Type of the document (e.g., CERTIFICATE) self.documentID = documentID # Unique ID for the document self.CarrierRMISID = carrierID # ID of the carrier associated with the document self.documentSizeInBytes = None self.documentTitle = None self.documentDate = None self.documentFileType = None self.documentData = None self.clientID = None class Delta: def __init__(self, insdID, timestamp): self.insdID = insdID # Carrier's RMIS ID self.timeStamp = timestamp # Timestamp used in the clear call ``` ```python TruckstopDBHelper.py import os import sqlite3 class DataBaseObject: def __init__(self, sqlConn, sqlCursor): self.sqlConn = sqlConn self.sqlCursor = sqlCursor def GetDBConnection(dbName:str): dbFilePath = "c:/TruckstopApp" os.makedirs(dbFilePath, exist_ok=True) sqlConn = sqlite3.connect(f'{dbFilePath}/{dbName}.sqlite') sqlCursor = sqlConn.cursor() dataBaseObject = DataBaseObject(sqlConn, sqlCursor) __CreateTables(dataBaseObject) return dataBaseObject def InsertIntoTable(table:str, dataBaseObject:DataBaseObject, fields:list[str]=None, values:list[str]=None, object:object = None, whereFields:list[str] = None, whereValues:list[str]=None): if object is not None: fields = list(vars(object).keys()) values = list(vars(object).values()) fields_string = ', '.join(fields) placeHolders = ', '.join(['?'] * len(values)) try: #Insert the data sqlQuery = f'INSERT INTO {table} ({fields_string}) values ({placeHolders})' dataBaseObject.sqlCursor.execute(sqlQuery, values) #Log the insert loqQuery = f'INSERT INTO CarrierLogs (ClientID, CarrierRMISID, Event, Message) values (?, ?, ?, ?)' logData = [object.clientID if object.clientID else None, object.CarrierRMISID if object.CarrierRMISID else None, f"Insert to {table}", f"Inserted {fields_string} with values {values}"] dataBaseObject.sqlCursor.execute(loqQuery, logData) dataBaseObject.sqlConn.commit() except Exception as e: # If there is already a record for this, update the record instead if 'UNIQUE constraint failed' in str(e) and table != "Documents": # Create the set clause dynamically for the UPDATE query update_fields = ', '.join([f"{field}=?" for field in fields]) if whereFields is not None: where_clause = " AND ".join([f"{whereField}=?" for whereField in whereFields]) else: where_clause = "" # Construct the UPDATE query with WHERE conditions for ClientID and CarrierRMISID sqlQuery = f'''UPDATE {table} SET {update_fields} WHERE {where_clause}''' update_values = values + (whereValues if whereValues is not None else []) #Insert the data dataBaseObject.sqlCursor.execute(sqlQuery, update_values) # Log the update loqQuery = f'INSERT INTO CarrierLogs (ClientID, Event, Message) values (?, ?, ?)' logData = [object.clientID, f"Update {table}", f"Updated {fields_string} with values {values}"] dataBaseObject.sqlCursor.execute(loqQuery, logData) dataBaseObject.sqlConn.commit() else: print (str(e)) def __CreateTables(dataBaseObject:DataBaseObject): dataBaseObject.sqlCursor.execute('''CREATE TABLE IF NOT EXISTS Documents ( ID INTEGER PRIMARY KEY, ClientID varchar(5), CarrierRMISID varchar(10), DocumentID varchar(10), DocumentType varchar(50), DocumentSizeInBytes varchar(50), DocumentTitle varchar(255), DocumentDate varchar(255), DocumentFileType varchar(10), DocumentData blob, UNIQUE(ClientID, DocumentID)) ''') dataBaseObject.sqlCursor.execute('''CREATE TABLE IF NOT EXISTS CarrierLogs ( ID INTEGER PRIMARY KEY, ClientID varchar(5), CarrierRMISID varchar(10), Event varchar(50), Message TEXT, Timestamp DATETIME DEFAULT CURRENT_TIMESTAMP) ''') dataBaseObject.sqlCursor.execute('''CREATE TABLE IF NOT EXISTS Carriers ( ID INTEGER PRIMARY KEY, ClientID varchar(5), CarrierName varchar(255), CarrierMC varchar(10), CarrierDOT varchar(50), CarrierRMISID varchar(10), LastUpdated varchar(50), UNIQUE(ClientID, CarrierRMISID)) ''') ``` # Extends Manage Onboarded Carrier Functionality Manage Onboarded Carriers # Overview **Flow:** # Retrieve Document from API **Flow:** # Process The API Response **Flow:**