diff --git a/emanifest-py/pyproject.toml b/emanifest-py/pyproject.toml index e434a85f..5c55136e 100644 --- a/emanifest-py/pyproject.toml +++ b/emanifest-py/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "emanifest" -version = "4.1.2" +version = "4.1.3" description = "An API utility wrapper for accessing the e-Manifest hazardous waste tracking system maintained by the US Environmental Protection Agency" readme = "README.md" authors = [ diff --git a/emanifest-py/requirements_dev.txt b/emanifest-py/requirements_dev.txt index 98bdaf56..c5273abb 100644 --- a/emanifest-py/requirements_dev.txt +++ b/emanifest-py/requirements_dev.txt @@ -3,3 +3,5 @@ ruff==0.5.2 responses==0.25.3 mypy==1.10.1 -r requirements.txt +typing_extensions==4.12.2 +types-requests==2.32.0.20240712 diff --git a/emanifest-py/src/emanifest/__init__.py b/emanifest-py/src/emanifest/__init__.py index 29d168b1..4a1a9d65 100644 --- a/emanifest-py/src/emanifest/__init__.py +++ b/emanifest-py/src/emanifest/__init__.py @@ -1,7 +1,33 @@ -from .client import ( # noqa: I001, F401 +from .client import ( + RCRAINFO_PREPROD, + RCRAINFO_PROD, RcrainfoClient, RcrainfoResponse, new_client, - RCRAINFO_PROD, - RCRAINFO_PREPROD, +) +from .types import ( + AdditionalInfo, + BillStatus, + BrInfo, + Comment, + CorrectionRequestStatus, + DotInformation, + Manifest, + MtnSearchArgs, + OriginType, + PcbInfo, + PortOfEntry, + Quantity, + RcraCodeDescription, + RcraLocality, + RcraSite, + SiteExistsResponse, + SiteSearchArgs, + SiteType, + Status, + SubmissionType, + UILinkArgs, + UserSearchArgs, + UserSearchResponse, + Waste, ) diff --git a/emanifest-py/src/emanifest/client.py b/emanifest-py/src/emanifest/client.py index 72ee601a..4ae3404f 100644 --- a/emanifest-py/src/emanifest/client.py +++ b/emanifest-py/src/emanifest/client.py @@ -2,14 +2,43 @@ e-Manifest library for using the e-Manifest API see https://github.com/USEPA/e-manifest """ + import io import json import zipfile from datetime import datetime, timezone -from typing import Any, Generic, List, Literal, Optional, TypedDict, TypeVar, Unpack +from typing import ( + Generic, + List, + Literal, + Optional, + TypeVar, +) from requests import Request, Response, Session -from requests_toolbelt.multipart import decoder, encoder +from requests_toolbelt.multipart import decoder, encoder # type: ignore +from typing_extensions import Unpack + +from .types import ( + AgencyCode, + CorrectionRevertResponse, + CorrectionVersionSearchArgs, + Manifest, + ManifestExistsResponse, + ManifestOperationResponse, + ManifestSignatureResponse, + MtnSearchArgs, + PortOfEntry, + RcraCodeDescription, + RcraSite, + SearchBillArgs, + SignManifestArgs, + SiteExistsResponse, + SiteSearchArgs, + UILinkArgs, + UserSearchArgs, + UserSearchResponse, +) RCRAINFO_PROD = "https://rcrainfo.epa.gov/rcrainfoprod/rest/api/" RCRAINFO_PREPROD = "https://rcrainfopreprod.epa.gov/rcrainfo/rest/api/" @@ -17,190 +46,6 @@ T = TypeVar("T") -class RcraCode(TypedDict): - """A code and an accompanying description""" - - code: str - description: str - - -class Phone(TypedDict): - number: str - - -class Contact(TypedDict): - firstName: str - lastName: str - phone: Phone - - -class Country(TypedDict): - code: str - name: str - - -class State(TypedDict): - code: str - name: str - - -class Address(TypedDict): - address1: str - city: str - country: Country - state: State - zip: str - - -SiteType = Literal["Generator", "Tsdf", "Transporter", "Broker", "RejectionInfo_AlternateTsdf"] - - -class RcraSite(TypedDict): - canEsign: bool - contact: Contact - epaSiteId: str - federalGeneratorStatus: str - gisPrimary: bool - hasRegisteredEmanifestUser: bool - limitedEsign: bool - mailingAddress: Address - name: str - siteAddress: Address - siteType: SiteType - - -class PortOfEntry(TypedDict): - """Ports that waste can enter/exit the US""" - - cityPort: str - state: State - - -class SiteExistsResponse(TypedDict): - """site exists service response""" - - epaSiteId: str - result: bool - - -class SiteSearchArgs(TypedDict, total=False): - epaSiteId: Optional[str] - name: Optional[str] - streetNumber: Optional[str] - address1: Optional[str] - city: Optional[str] - state: Optional[str] - zip: Optional[str] - siteType: Optional[str] - pageNumber: Optional[int] - - -class UserSearchArgs(TypedDict, total=False): - userId: Optional[str] - siteIds: Optional[List[str]] - pageNumber: Optional[int] - - -class UserPermission(TypedDict): - """A user's permissions for a module in RCRAInfo for a given site""" - - level: str - module: str - - -class UserSite(TypedDict): - """A user's permissions for a site""" - - permissions: List[UserPermission] - siteId: str - siteName: str - - -class User(TypedDict): - """User details from the user search service""" - - email: str - esaStatus: str - firstName: str - lastLoginDate: str - lastName: str - phone: Phone - sites: List[UserSite] - userId: str - - -class UserSearchResponse(TypedDict): - """body of the response from the user search service""" - - currentPageNumber: int - searchedParameters: list - totalNumberOfPages: int - totalNumberOfUsers: int - users: List[User] - warnings: List[Any] - - -CorrectionRequestStatus = Literal["NotSent", "Sent", "IndustryResponded", "Cancelled"] - -DateType = Literal["CertifiedDate", "ReceivedDate", "ShippedDate", "UpdatedDate", "QuickSignDate"] - -SubmissionType = Literal["FullElectronic", "Hybrid", "Image", "DataImage5Copy"] - -Status = Literal[ - "Pending", - "Scheduled", - "InTransit", - "Received", - "ReadyForSignature", - "Signed", - "SignedComplete", - "UnderCorrection", - "Corrected", -] - - -class ManifestComment(TypedDict): - """A comment on a manifest""" - - label: str - description: str - handlerId: str - - -class MtnSearchParams(TypedDict, total=False): - """Search parameters for manifest tracking numbers""" - - stateCode: str - siteId: str - submissionType: SubmissionType - status: Status - dateType: DateType - siteType: SiteType - transporterOrder: int - startDate: datetime - endDate: datetime - correctionRequestStatus: CorrectionRequestStatus - comments: ManifestComment - - -class CorrectionVersionSearchParams(TypedDict, total=False): - """Search parameters for manifest correction versions""" - - manifestTrackingNumber: str - status: Status - ppcStatus: Literal["PendingDataEntry", "DataQaCompleted"] - versionNumber: str - - -class CorrectionRevert(TypedDict): - """Information returned after successfully reverting a manifest under correction""" - - currentVersionNumber: int - date: str - manifestTrackingNumber: str - operationStatus: str - - class RcrainfoResponse(Generic[T]): """ RcrainfoResponse wraps around the requests library's Response object. @@ -210,9 +55,9 @@ class RcrainfoResponse(Generic[T]): response (Response) the request library response object. """ - def __init__(self, response): + def __init__(self, response: Response): self.response: Response = response - self._multipart_json: T = None + self._multipart_json: T | None = None self._multipart_zip: Optional[zipfile.ZipFile] = None def json(self) -> T: @@ -268,7 +113,9 @@ class RcrainfoClient(Session): """ # see datetime docs https://docs.python.org/3.11/library/datetime.html#strftime-strptime-behavior + # acceptable date format(s) [yyyy-MM-dd'T'HH:mm:ssZ,yyyy-MM-dd'T'HH:mm:ss.SSSZ] __expiration_fmt = "%Y-%m-%dT%H:%M:%S.%f%z" + __signature_date_fmt = "%Y-%m-%dT%H:%M:%SZ" __default_headers = {"Accept": "application/json"} __default_timeout = 10 @@ -279,7 +126,7 @@ def __init__( self.base_url = base_url self.timeout = timeout self.__token = None - self.__token_expiration_utc = datetime.utcnow().replace(tzinfo=timezone.utc) + self.__token_expiration_utc = datetime.now(timezone.utc).replace(tzinfo=timezone.utc) self.__api_key = api_key self.__api_id = api_id self.headers.update(self.__default_headers) @@ -287,7 +134,7 @@ def __init__( @property def base_url(self): - """RCRAInfo base URL, either for Production ('prod') or Preproduction ('preprod') for testing.""" + """RCRAInfo base URL, either for Production ('prod') or Preproduction ('preprod')""" return self.__base_url @base_url.setter @@ -317,7 +164,7 @@ def __set_token_expiration(self, expiration: str) -> None: expiration, self.__expiration_fmt ).replace(tzinfo=timezone.utc) except ValueError: - self.__token_expiration_utc = datetime.utcnow().replace(tzinfo=timezone.utc) + self.__token_expiration_utc = datetime.now(timezone.utc).replace(tzinfo=timezone.utc) @property def token(self): @@ -329,7 +176,7 @@ def is_authenticated(self) -> bool: """Returns True if the RcrainfoClient token exists and has not expired.""" try: if ( - self.token_expiration > datetime.utcnow().replace(tzinfo=timezone.utc) + self.token_expiration > datetime.now(timezone.utc).replace(tzinfo=timezone.utc) and self.token is not None ): return True @@ -398,7 +245,7 @@ def __get_token(self) -> None: @staticmethod def __encode_manifest( manifest_json: dict, - zip_bytes: bytes = None, + zip_bytes: bytes | None = None, *, json_name: str = "manifest.json", zip_name="attachments.zip", @@ -426,7 +273,6 @@ def __encode_manifest( } ) - # The following methods are exposed so users can hook into our client and customize its behavior def retrieve_id(self, api_id=None) -> str: """ Getter method used internally to retrieve the desired RCRAInfo API ID. Can be overridden @@ -441,27 +287,19 @@ def retrieve_id(self, api_id=None) -> str: return api_id elif self.__api_id: return self.__api_id - elif self.__api_id is None and api_id is None: - # pass an empty string so, if user's fail to provide a string, it will not try to use None as - # the API credential in the URL + else: return "" - def retrieve_key(self, api_key=None) -> str: + def retrieve_key(self, api_key: str | None = None) -> str: """ Getter method used internally to retrieve the desired RCRAInfo API key. Can be overridden - to automatically support retrieving an API Key from an external location. - - Args: - api_key: - - Returns: - string of the user's RCRAInfo API Key + to support retrieving an API Key from an external location. """ if api_key: return api_key elif self.__api_key: return self.__api_key - elif self.__api_key is None and api_key is None: + else: return "" # Below this line are the high level methods to request RCRAInfo/e-Manifest @@ -472,9 +310,6 @@ def authenticate(self, api_id=None, api_key=None) -> None: Args: api_id (str): API ID of RCRAInfo User with Site Manager level permission. api_key (str): User's RCRAInfo API key. Generated alongside the api_id in RCRAInfo - - Returns: - token (client): Authentication token for use by other emanifest functions. Expires after 20 minutes """ # if api credentials are passed, set the client's attributes if api_id is not None: @@ -553,7 +388,7 @@ def get_ship_name_by_id(self, id_num: str) -> RcrainfoResponse[List[str]]: ) return self.__rcra_request("GET", endpoint) - def get_mtn_suffix(self) -> RcrainfoResponse[List[RcraCode]]: + def get_mtn_suffix(self) -> RcrainfoResponse[List[RcraCodeDescription]]: """Retrieve Allowable Manifest Tracking Number (MTN) Suffixes""" endpoint = f"{self.base_url}v1/emanifest/lookup/printed-tracking-number-suffixes" return self.__rcra_request("GET", endpoint) @@ -573,14 +408,14 @@ def get_container_types(self) -> RcrainfoResponse: endpoint = f"{self.base_url}v1/emanifest/lookup/container-types" return self.__rcra_request("GET", endpoint) - def get_quantity_uom(self) -> RcrainfoResponse[List[RcraCode]]: + def get_quantity_uom(self) -> RcrainfoResponse[List[RcraCodeDescription]]: """ Retrieve Quantity Units of Measure (UOM) """ endpoint = f"{self.base_url}v1/emanifest/lookup/quantity-uom" return self.__rcra_request("GET", endpoint) - def get_load_types(self) -> RcrainfoResponse[List[RcraCode]]: + def get_load_types(self) -> RcrainfoResponse[List[RcraCodeDescription]]: """Retrieve PCB Load Types""" endpoint = f"{self.base_url}v1/emanifest/lookup/load-types" return self.__rcra_request("GET", endpoint) @@ -600,22 +435,24 @@ def get_id_numbers(self) -> RcrainfoResponse[List[str]]: endpoint = f"{self.base_url}v1/emanifest/lookup/id-numbers" return self.__rcra_request("GET", endpoint) - def get_density_uom(self) -> RcrainfoResponse[List[RcraCode]]: + def get_density_uom(self) -> RcrainfoResponse[List[RcraCodeDescription]]: """Retrieve Density Units of Measure (UOM)""" endpoint = f"{self.base_url}v1/lookup/density-uom" return self.__rcra_request("GET", endpoint) - def get_form_codes(self) -> RcrainfoResponse[list[RcraCode]]: + def get_form_codes(self) -> RcrainfoResponse[list[RcraCodeDescription]]: """Retrieve Form Codes""" endpoint = f"{self.base_url}v1/lookup/form-codes" return self.__rcra_request("GET", endpoint) - def get_source_codes(self) -> RcrainfoResponse[List[RcraCode]]: + def get_source_codes(self) -> RcrainfoResponse[List[RcraCodeDescription]]: """Retrieve Source Codes""" endpoint = f"{self.base_url}v1/lookup/source-codes" return self.__rcra_request("GET", endpoint) - def get_state_waste_codes(self, state_code: str) -> RcrainfoResponse[List[RcraCode]]: + def get_state_waste_codes( + self, state_code: str + ) -> RcrainfoResponse[List[RcraCodeDescription]]: """ Retrieve State Waste Codes for a given state (besides Texas) @@ -625,17 +462,17 @@ def get_state_waste_codes(self, state_code: str) -> RcrainfoResponse[List[RcraCo endpoint = f"{self.base_url}v1/lookup/state-waste-codes/{state_code}" return self.__rcra_request("GET", endpoint) - def get_fed_waste_codes(self) -> RcrainfoResponse[List[RcraCode]]: + def get_fed_waste_codes(self) -> RcrainfoResponse[List[RcraCodeDescription]]: """Retrieve Federal Waste Codes""" endpoint = f"{self.base_url}v1/lookup/federal-waste-codes" return self.__rcra_request("GET", endpoint) - def get_man_method_codes(self) -> RcrainfoResponse[List[RcraCode]]: + def get_man_method_codes(self) -> RcrainfoResponse[List[RcraCodeDescription]]: """Retrieve Management Method Codes""" endpoint = f"{self.base_url}v1/lookup/management-method-codes" return self.__rcra_request("GET", endpoint) - def get_waste_min_codes(self) -> RcrainfoResponse[List[RcraCode]]: + def get_waste_min_codes(self) -> RcrainfoResponse[List[RcraCodeDescription]]: """Retrieve Waste Minimization Codes""" endpoint = f"{self.base_url}v1/lookup/waste-minimization-codes" return self.__rcra_request("GET", endpoint) @@ -700,36 +537,18 @@ def get_bill(self, **kwargs) -> RcrainfoResponse: endpoint = f"{self.base_url}v1/emanifest/billing/bill" return self.__rcra_request("POST", endpoint, **kwargs) - def search_bill(self, **kwargs) -> RcrainfoResponse: - """ - Search and retrieve bills using all or some of the provided criteria - - Keyword Args: - billingAccount (str): EPA Site ID - billStatus (str): Active, Paid, Unpaid, ReadyForPayment, Credit, InProgress, SendToCollections, ZeroBalance. - startDate(date): Beginning of the billing period (yyyy-MM-dd'T'HH:mm:ssZ or yyyy-MM-dd'T'HH:mm:ss.SSSZ) - endDate (date): End of the billing period (yyyy-MM-dd'T'HH:mm:ssZ or yyyy-MM-dd'T'HH:mm:ss.SSSZ) - amountChanged (boolean): True or false - pageNumber (number): Must be greater than 0 - - Returns: - dict: object with bills matching criteria - """ + def search_bill(self, **kwargs: Unpack[SearchBillArgs]) -> RcrainfoResponse: + """Search and retrieve bills using all or some of the provided criteria""" endpoint = f"{self.base_url}v1/emanifest/billing/bill-search" return self.__rcra_request("POST", endpoint, **kwargs) - def get_manifest_attachments(self, mtn: str, reg: bool = False) -> RcrainfoResponse: + def get_manifest_attachments(self, mtn: str, reg: bool = False) -> RcrainfoResponse[Manifest]: """ - Retrieve e-Manifest details as json with attachments matching provided Manifest Tracking Number (MTN) - - Args: - mtn (str): Manifest tracking number - reg (bool): use endpoint for regulators, defaults to False + Retrieve manifest data (JSON) and attachments Returns: json: Downloaded file containing e-Manifest details for given MTN - attachments: PDF and HTML files containing additional manifest information (such as scans - or electronic copies) for the given MTN + attachments: PDF and HTML files (such as scans or electronic copies) """ if reg: endpoint = f"{self.base_url}v1/state/emanifest/manifest/{mtn}/attachments" @@ -743,27 +562,11 @@ def get_manifest_attachments(self, mtn: str, reg: bool = False) -> RcrainfoRespo return resp def search_mtn( - self, reg: bool = False, **kwargs: Unpack[MtnSearchParams] + self, reg: bool = False, **kwargs: Unpack[MtnSearchArgs] ) -> RcrainfoResponse[List[str]]: """ Retrieve manifest tracking numbers based on all or some of provided search criteria - Args: - reg (bool): Use the Regulator endpoint, defaults to False - - Keyword Args: - stateCode (str): Two-letter US postal state code - siteId (str): EPA Site ID - submissionType (str): FullElectronic, Hybrid, Image, DataImage5Copy - status (Status): manifest status - dateType (DateType): Type of date to filter MTN by. Case-sensitive - siteType (SiteType): Type of site/handler to filter MTN by. Case-sensitive - transporterOrder (int): Number representing the order of a transporter on the manifest - startDate (date): start of range (yyyy-MM-dd'T'HH:mm:ssZ or yyyy-MM-dd'T'HH:mm:ss.SSSZ) - endDate (date): End of range (yyyy-MM-dd'T'HH:mm:ssZ or yyyy-MM-dd'T'HH:mm:ss.SSSZ) - correctionRequestStatus (CorrectionRequestStatus): correction status - comments (ManifestComment): filter manifest by comments - Returns: list: list of manifest tracking numbers """ @@ -773,13 +576,9 @@ def search_mtn( endpoint = f"{self.base_url}v1/emanifest/search" return self.__rcra_request("POST", endpoint, **kwargs) - def get_correction(self, mtn: str, reg: bool = False) -> RcrainfoResponse: + def get_correction(self, mtn: str, reg: bool = False) -> RcrainfoResponse[Manifest]: """ - Retrieve information about all manifest correction versions by manifest tracking number (MTN) - - Args: - mtn (str): Manifest tracking number - reg (bool): use endpoint for regulators, defaults to False + Retrieve manifest correction version details by manifest tracking number (MTN) Returns: dict: object containing correction details for given MTN @@ -791,8 +590,8 @@ def get_correction(self, mtn: str, reg: bool = False) -> RcrainfoResponse: return self.__rcra_request("GET", endpoint) def get_correction_version( - self, reg: bool = False, **kwargs: Unpack[CorrectionVersionSearchParams] - ) -> RcrainfoResponse: # ToDo: return type is manifest + self, reg: bool = False, **kwargs: Unpack[CorrectionVersionSearchArgs] + ) -> RcrainfoResponse[Manifest]: """ Retrieve details of manifest correction version based the provided search criteria @@ -805,24 +604,22 @@ def get_correction_version( endpoint = f"{self.base_url}v1/emanifest/manifest/correction-version" return self.__rcra_request("POST", endpoint, **kwargs) - def get_correction_attachments(self, **kwargs) -> RcrainfoResponse: + def get_correction_attachments(self, **kwargs) -> RcrainfoResponse[Manifest]: """ - Retrieve attachments of corrected manifests based all or some of the provided search criteria. + Retrieve corrected manifests attachment by search criteria. See also **get_correction** and **get_correction_version** Keyword Args: manifestTrackingNumber (str): Manifest tracking number. Required status (str): Manifest status (Signed, Corrected, UnderCorrection). Case-sensitive - ppcStatus (str): EPA Paper Processing Center Status (PendingDataEntry, DataQaCompleted). Case-sensitive + ppcStatus (str): EPA Paper Processing Center Status (PendingDataEntry, DataQaCompleted) versionNumber (str): Manifest version number Returns: json: Downloaded file containing e-Manifest details for given MTN attachments: PDF and HTML files containing additional - manifest information (such as scans or - electronic copies) for the given MTN - print: message of success or failure + manifest information (such as scans or copies) for the given MTN """ endpoint = f"{self.base_url}v1/emanifest/manifest/correction-version/attachments" resp = self.__rcra_request("POST", endpoint, **kwargs) @@ -831,29 +628,15 @@ def get_correction_attachments(self, **kwargs) -> RcrainfoResponse: return resp def get_site_mtn(self, site_id: str, reg: bool = False) -> RcrainfoResponse[List[str]]: - """ - Retrieve manifest tracking numbers for a given Site ID - - Args: - site_id (str): EPA Site ID - reg (bool): use endpoint for regulators, defaults to False - """ + """Retrieve manifest tracking numbers for a given Site ID""" if reg: endpoint = f"{self.base_url}v1/state/emanifest/manifest-tracking-numbers/{site_id}" else: endpoint = f"{self.base_url}v1/emanifest/manifest-tracking-numbers/{site_id}" return self.__rcra_request("GET", endpoint) - def get_manifest( - self, mtn: str, reg: bool = False - ) -> RcrainfoResponse: # ToDO: return type is manifest - """ - Retrieve e-Manifest details matching provided Manifest Tracking Number (MTN) - - Args: - mtn (str): Manifest tracking number - reg (bool): use endpoint for regulators, defaults to False - """ + def get_manifest(self, mtn: str, reg: bool = False) -> RcrainfoResponse[Manifest]: + """Retrieve e-Manifest details by Manifest Tracking Number (MTN)""" if reg: endpoint = f"{self.base_url}v1/state/emanifest/manifest/{mtn}" else: @@ -877,7 +660,9 @@ def get_sites( endpoint = f"{self.base_url}v1/emanifest/site-ids/{state_code}/{site_type}" return self.__rcra_request("GET", endpoint) - def correct_manifest(self, manifest_json: dict, zip_file: bytes = None) -> RcrainfoResponse: + def correct_manifest( + self, manifest_json: dict, zip_file: bytes | None = None + ) -> RcrainfoResponse[ManifestOperationResponse]: """ Correct Manifest by providing e-Manifest JSON and optional Zip attachment @@ -895,20 +680,19 @@ def correct_manifest(self, manifest_json: dict, zip_file: bytes = None) -> Rcrai endpoint = f"{self.base_url}v1/emanifest/manifest/correct" return self.__rcra_request("PUT", endpoint, multipart=multipart) - def revert_manifest(self, mtn: str) -> RcrainfoResponse[CorrectionRevert]: + def revert_manifest(self, mtn: str) -> RcrainfoResponse[CorrectionRevertResponse]: """ Revert manifest in 'UnderCorrection' status to previous 'Corrected' or 'Signed' version - Args: - mtn (str): Manifest tracking number - Returns: dict: object containing confirmation of correction """ endpoint = f"{self.base_url}v1/emanifest/manifest/revert/{mtn}" return self.__rcra_request("GET", endpoint) - def patch_update_manifest(self, mtn: str, data: dict) -> RcrainfoResponse: + def patch_update_manifest( + self, mtn: str, data: dict + ) -> RcrainfoResponse[ManifestOperationResponse]: """ Update a portion of a manifest via the patch process @@ -924,7 +708,9 @@ def patch_update_manifest(self, mtn: str, data: dict) -> RcrainfoResponse: "PATCH", endpoint, headers={"Content-Type": "application/json-patch+json"}, **data ) - def patch_correct_manifest(self, mtn: str, data: dict) -> RcrainfoResponse: + def patch_correct_manifest( + self, mtn: str, data: dict + ) -> RcrainfoResponse[ManifestOperationResponse]: """ Update a portion of a manifest via the patch process @@ -940,7 +726,9 @@ def patch_correct_manifest(self, mtn: str, data: dict) -> RcrainfoResponse: "PATCH", endpoint, headers={"Content-Type": "application/json-patch+json"}, **data ) - def update_manifest(self, manifest_json: dict, zip_file: bytes = None) -> RcrainfoResponse: + def update_manifest( + self, manifest_json: dict, zip_file: bytes | None = None + ) -> RcrainfoResponse[ManifestOperationResponse]: """ Update Manifest by providing e-Manifest JSON and optional Zip attachment @@ -966,43 +754,33 @@ def update_manifest(self, manifest_json: dict, zip_file: bytes = None) -> Rcrain multipart=multipart, ) - def delete_manifest(self, mtn: str) -> RcrainfoResponse: + def delete_manifest(self, mtn: str) -> RcrainfoResponse[ManifestOperationResponse]: """ Delete selected manifest - Args: - mtn (str): Manifest tracking number - Returns: dict: message of success or failure """ endpoint = f"{self.base_url}v1/emanifest/manifest/delete/{mtn}" return self.__rcra_request("DELETE", endpoint) - def sign_manifest(self, **kwargs) -> RcrainfoResponse: - """ - Quicker sign selected manifests - - Keyword Args: - manifestTrackingNumbers (array) : An array of manifest tracking numbers to sign - siteId (str) : The EPA ID for the site that signs - siteType (str) : The site on the manifest that is signing (Generator, Tsdf, Transporter, RejectionInfo_AlternateTsdf). Case-sensitive - printedSignatureName (str) : The name of the person who signed the manifest - printedSignatureDate (date) : The date the person signed the manifest - transporterOrder (int) : If the site is a transporter, the order of that transporter on the manifest - - Returns: - dict: message of success or failure - """ + def sign_manifest( + self, **kwargs: Unpack[SignManifestArgs] + ) -> RcrainfoResponse[ManifestSignatureResponse]: + """Quicker sign selected manifests""" + sign_date = kwargs.get("printedSignatureDate", None) + if not sign_date: + kwargs["printedSignatureDate"] = datetime.now(timezone.utc).strftime( + self.__signature_date_fmt + ) + if isinstance(sign_date, datetime): + kwargs["printedSignatureDate"] = sign_date.strftime(self.__signature_date_fmt) endpoint = f"{self.base_url}v1/emanifest/manifest/quicker-sign" return self.__rcra_request("POST", endpoint, **kwargs) def get_available_manifests(self, mtn: str) -> RcrainfoResponse: """ - Returns previous and future signature-related information about manifest and respective sites - - Args: - mtn (str): Manifest tracking number + Returns signature-related information about manifest and respective sites Returns: dict: object containing manifest signature details @@ -1010,7 +788,9 @@ def get_available_manifests(self, mtn: str) -> RcrainfoResponse: endpoint = f"{self.base_url}v1/emanifest/manifest/available-to-sign/{mtn}" return self.__rcra_request("GET", endpoint) - def save_manifest(self, manifest_json: dict, zip_file: bytes = None) -> RcrainfoResponse: + def save_manifest( + self, manifest_json: dict, zip_file: bytes | None = None + ) -> RcrainfoResponse[ManifestOperationResponse]: """ Save Manifest by providing e-Manifest JSON and optional Zip attachment @@ -1036,30 +816,15 @@ def save_manifest(self, manifest_json: dict, zip_file: bytes = None) -> Rcrainfo multipart=multipart, ) - def check_mtn_exists(self, mtn: str) -> RcrainfoResponse: - """ - Check if Manifest Tracking Number (MTN) exists and return basic details - - Args: - mtn (str): Manifest tracking number - - Returns: - dict: object containing MTN details and confirmation if site exists - """ + def check_mtn_exists(self, mtn: str) -> RcrainfoResponse[ManifestExistsResponse]: + """Check if Manifest Tracking Number (MTN) exists and return basic details""" endpoint = f"{self.base_url}v1/emanifest/manifest/mtn-exists/{mtn}" return self.__rcra_request("GET", endpoint) - def get_ui_link(self, **kwargs) -> RcrainfoResponse: + def get_ui_link(self, **kwargs: Unpack[UILinkArgs]) -> RcrainfoResponse: """ Generate link to the user interface (UI) of the RCRAInfo e-Manifest module - Keyword Args: - page (str): Dashboard, BulkSign, BulkQuickSign, Edit, View, Sign. Case-sensitive - epaSiteId (str): EPA Site ID - manifestTrackingNumber (str): Manifest tracking number (optional) - filter (list): List of MTNs (optional) - view (str) : Incoming, Outgoing, All, Transporting, Broker, CorrectionRequests, Original, Corrections - Returns: dict: object containing link to UI """ @@ -1067,7 +832,7 @@ def get_ui_link(self, **kwargs) -> RcrainfoResponse: return self.__rcra_request("POST", endpoint, **kwargs) def get_cme_lookup( - self, activity_location: str, agency_code: str, nrr_flag: bool = True + self, activity_location: str, agency_code: AgencyCode, nrr_flag: bool = True ) -> RcrainfoResponse: """ Retrieve all lookups for specific activity location and agency code, including staff, @@ -1075,16 +840,7 @@ def get_cme_lookup( Args: activity_location (str): Two-letter US postal state code - agency_code (str): One-letter code. B (State Contractor/Grantee), - C (EPA Contractor/Grantee), - E (EPA), - L (Local), - N (Native American), - S (State), - T (State-Initiated Oversight/Observation/Training Actions), - X (EPA-Initiated Oversight/Observation/Training Actions), - J (Joint State), - P (Joint EPA) + agency_code (AgencyCode): One-letter code. nrr_flag (boolean): True/False if Non-Financial Record Review Returns: @@ -1123,7 +879,7 @@ def get_handler(self, handler_id: str, details: bool = False) -> RcrainfoRespons Args: handler_id (str): EPA Site ID number - details (boolean): True/false to request additional details. Optional; defaults to False + details (boolean): True/false to request additional details. Optional; defaults False Returns: dict: object containing handler source records (and optional details) @@ -1146,10 +902,13 @@ def _parse_url(base_url: str | None) -> str: return base_url +BaseUrls = Literal["prod", "preprod"] + + def new_client( - base_url: str = None, - api_id: str = None, - api_key: str = None, + base_url: BaseUrls | str | None = None, + api_id: str | None = None, + api_key: str | None = None, auto_renew: bool = False, ) -> RcrainfoClient: """ @@ -1164,4 +923,6 @@ def new_client( Returns: RcrainfoClient: RCRAInfo client instance """ + if base_url is None: + raise ValueError("base_url is required") return RcrainfoClient(base_url, api_id=api_id, api_key=api_key, auto_renew=auto_renew) diff --git a/emanifest-py/src/emanifest/test_client.py b/emanifest-py/src/emanifest/test_client.py index b2896c71..121b5667 100644 --- a/emanifest-py/src/emanifest/test_client.py +++ b/emanifest-py/src/emanifest/test_client.py @@ -4,7 +4,7 @@ import requests import responses -from . import RCRAINFO_PREPROD, RcrainfoClient, RcrainfoResponse, new_client +from . import RCRAINFO_PREPROD, RcrainfoClient, new_client MOCK_MTN = "100032437ELC" MOCK_GEN_ID = "VATESTGEN001" @@ -71,7 +71,7 @@ def test_extracted_response_json_matches(self, mock_response): }, ) # Act - resp: RcrainfoResponse = rcrainfo.get_site(MOCK_GEN_ID) + resp = rcrainfo.get_site(MOCK_GEN_ID) # Assert assert resp.response.json() == resp.json() @@ -84,7 +84,7 @@ class MyClass(RcrainfoClient): def retrieve_id(self, api_id=None) -> str: """ This example method on our test subclass shows we can override the set_api_id method - if the user wants to get their api ID from somewhere else (e.g., a service, or database) + if the user wants to get their api ID from somewhere else (e.g., a service, database) """ returned_string = ( self.mock_api_id_from_external @@ -216,11 +216,13 @@ class TestNewClientConstructor: def test_returns_instance_of_client(self): rcrainfo = new_client("prod") preprod = new_client("preprod") - blank = new_client() assert isinstance(rcrainfo, RcrainfoClient) assert isinstance(preprod, RcrainfoClient) - assert isinstance(blank, RcrainfoClient) def test_new_client_defaults_to_preprod(self): - rcrainfo = new_client() + rcrainfo = new_client("preprod") assert rcrainfo.base_url == RCRAINFO_PREPROD + + def test_no_base_url_raises_exception(self): + with pytest.raises(ValueError): + new_client() diff --git a/emanifest-py/src/emanifest/types.py b/emanifest-py/src/emanifest/types.py new file mode 100644 index 00000000..4fe50fb0 --- /dev/null +++ b/emanifest-py/src/emanifest/types.py @@ -0,0 +1,450 @@ +from datetime import datetime +from typing import ( + Any, + List, + Literal, + Optional, + TypedDict, + Union, +) + +from typing_extensions import Required + + +class RcraCodeDescription(TypedDict, total=False): + """A code and an accompanying description""" + + code: Required[str] + description: str + + +class Phone(TypedDict, total=False): + """A RCRAInfo phone number""" + + number: Required[str] + extension: Optional[str] + + +class Contact(TypedDict, total=False): + """RCRAInfo user contact information""" + + firstName: str + middleInitial: str + lastName: str + phone: Phone + email: str + companyName: str + + +class RcraLocality(TypedDict, total=False): + """A RCRAInfo locality (state or country)""" + + code: Required[str] + name: str + + +class Address(TypedDict, total=False): + """An address in the RCRAInfo system""" + + address1: Required[str] + address2: str + city: str + country: RcraLocality + state: RcraLocality + zip: str + + +SiteType = Literal["Generator", "Tsdf", "Transporter", "Broker", "RejectionInfo_AlternateTsdf"] + + +class RcraSite(TypedDict): + """A site in the RCRAInfo system""" + + canEsign: bool + contact: Contact + epaSiteId: str + federalGeneratorStatus: str + gisPrimary: bool + hasRegisteredEmanifestUser: bool + limitedEsign: bool + mailingAddress: Address + name: str + siteAddress: Address + siteType: SiteType + + +class PortOfEntry(TypedDict): + """Ports that waste can enter/exit the US""" + + cityPort: str + state: RcraLocality + + +class SiteExistsResponse(TypedDict): + """site exists service response""" + + epaSiteId: str + result: bool + + +class SiteSearchArgs(TypedDict, total=False): + """Search parameters for site search service""" + + epaSiteId: Optional[str] + name: Optional[str] + streetNumber: Optional[str] + address1: Optional[str] + city: Optional[str] + state: Optional[str] + zip: Optional[str] + siteType: Optional[str] + pageNumber: Optional[int] + + +class UserSearchArgs(TypedDict, total=False): + """Search parameters for user search service""" + + userId: Optional[str] + siteIds: Optional[List[str]] + pageNumber: Optional[int] + + +class _UserPermission(TypedDict): + """A user's permissions for a module in RCRAInfo for a given site""" + + level: str + module: str + + +class _UserSiteAccess(TypedDict): + """A user's permissions for a site""" + + permissions: List[_UserPermission] + siteId: str + siteName: str + + +class _UserWithSiteAccess(TypedDict): + """User details from the user search service""" + + email: str + esaStatus: str + firstName: str + lastLoginDate: str + lastName: str + phone: Phone + sites: List[_UserSiteAccess] + userId: str + + +class UserSearchResponse(TypedDict): + """body of the response from the user search service""" + + currentPageNumber: int + searchedParameters: list + totalNumberOfPages: int + totalNumberOfUsers: int + users: List[_UserWithSiteAccess] + warnings: List[Any] + + +CorrectionRequestStatus = Literal["NotSent", "Sent", "IndustryResponded", "Cancelled"] + +DateType = Literal["CertifiedDate", "ReceivedDate", "ShippedDate", "UpdatedDate", "QuickSignDate"] + +SubmissionType = Literal["FullElectronic", "Hybrid", "Image", "DataImage5Copy"] + +Status = Literal[ + "Pending", + "Scheduled", + "InTransit", + "Received", + "ReadyForSignature", + "Signed", + "SignedComplete", + "UnderCorrection", + "Corrected", +] + + +class _ManifestComment(TypedDict): + """A comment on a manifest""" + + label: str + description: str + handlerId: str + + +class MtnSearchArgs(TypedDict, total=False): + """Search parameters for manifest tracking numbers""" + + stateCode: str + siteId: str + submissionType: SubmissionType + status: Status + dateType: DateType + siteType: SiteType + transporterOrder: int + startDate: datetime + endDate: datetime + correctionRequestStatus: CorrectionRequestStatus + comments: _ManifestComment + + +BillStatus = Literal[ + "Active", + "Paid", + "Unpaid", + "ReadyForPayment", + "Credit", + "InProgress", + "SendToCollections", + "ZeroBalance", +] + + +class SearchBillArgs(TypedDict, total=False): + """Search parameters for billing history""" + + billingAccount: str + billStatus: BillStatus + startDate: datetime | str + endDate: datetime | str + amountChanged: bool + pageNumber: int + + +class CorrectionVersionSearchArgs(TypedDict, total=False): + """Search parameters for manifest correction versions""" + + manifestTrackingNumber: str + status: Status + ppcStatus: Literal["PendingDataEntry", "DataQaCompleted"] + versionNumber: str + + +class CorrectionRevertResponse(TypedDict): + """Information returned after successfully reverting a manifest under correction""" + + currentVersionNumber: int + date: str + manifestTrackingNumber: str + operationStatus: str + + +class ManifestOperationResponse(TypedDict): + """Information returned to indicate the status of a manifest operation""" + + manifestTrackingNumber: str + operationStatus: str + date: datetime + + +class SignManifestArgs(TypedDict, total=False): + """Arguments for Quick signing a manifest through the quicker-sign service""" + + manifestTrackingNumbers: Required[List[str]] + siteId: Required[str] + siteType: Required[SiteType] + printedSignatureName: Required[str] + printedSignatureDate: Required[Union[datetime, str]] + transporterOrder: int + + +class _QuickSignManifestReport(TypedDict): + """Part of body returned from the quicker-sign service""" + + manifestTrackingNumber: str + + +class _QuickSignSignerReport(TypedDict, total=False): + """Report of the signer's actions. Part of body returned from the quicker-sign service""" + + electronicSignatureDate: datetime + firstName: str + lastName: str + printedSignatureDate: datetime + printedSignatureName: str + userId: str + warnings: List[dict] + + +class _QuickSignSiteReport(TypedDict): + """Part of body returned from the quicker-sign service""" + + siteId: str + siteType: str + + +class ManifestSignatureResponse(ManifestOperationResponse): + """Status report returned from the quicker-sign service""" + + manifestReports: List[_QuickSignManifestReport] + reportId: str + signerReport: _QuickSignSignerReport + siteReport: _QuickSignSiteReport + + +UiLinkViews = Literal[ + "Incoming", + "Outgoing", + "All", + "Transporting", + "Broker", + "CorrectionRequests", + "Original", + "Corrections", +] + +UiLinkPages = Literal["Dashboard", "BulkSign", "BulkQuickSign", "Edit", "View", "Sign"] + + +class UILinkArgs(TypedDict, total=False): + """Arguments for generating a Manifest UI link""" + + page: Required[UiLinkPages] + epaSiteId: Required[str] + manifestTrackingNumber: str + filter: List[str] + view: UiLinkViews + + +AgencyCode = Literal["B", "C", "E", "L", "N", "S", "T", "X", "J", "P"] + +OriginType = Literal["Service", "Web", "Mail"] + + +class DiscrepancyResidueInfo(TypedDict, total=False): + """Manifest Waste Discrepancy info""" + + wasteQuantity: bool + wasteType: bool + discrepancyComments: Optional[str] + residue: bool + residueComments: Optional[str] + + +class DotInformation(TypedDict): + """Waste Line Department of Transportation Information""" + + idNumber: RcraCodeDescription + printedDotInformation: str + + +class Quantity(TypedDict, total=False): + """Waste Line Quantities""" + + containerNumber: int + containerType: RcraCodeDescription + quantity: int + unitOfMeasurement: RcraCodeDescription + + +class BrInfo(TypedDict, total=False): + """Biennial Report Information that can be added to a manifest""" + + density: Optional[float] + densityUnitOfMeasurement: Optional[RcraCodeDescription] + formCode: Optional[RcraCodeDescription] + sourceCode: Optional[RcraCodeDescription] + wasteMinimizationCode: Optional[RcraCodeDescription] + mixedRadioactiveWaste: Optional[bool] + + +class HazardousWaste(TypedDict, total=False): + """Key codes that indicate the type of hazardous waste""" + + federalWasteCodes: List[RcraCodeDescription] + tsdfStateWasteCodes: List[RcraCodeDescription] + txWasteCodes: List[str] + generatorStateWasteCodes: List[RcraCodeDescription] + + +class Comment(TypedDict, total=False): + """A comment on a manifest or waste line""" + + label: Optional[str] + description: Optional[str] + handlerId: Optional[str] + + +class AdditionalInfo(TypedDict, total=False): + """Additional Information structure for use on a manifest or individual waste line""" + + originalManifestTrackingNumbers: Optional[List[str]] + newManifestDestination: Optional[str] + consentNumber: Optional[str] + comments: Optional[List[Comment]] + + +class PcbInfo(TypedDict, total=False): + """Polychlorinated biphenyls information on a manifest waste line""" + + loadType: Optional[RcraCodeDescription] + articleContainerId: Optional[str] + dateOfRemoval: Optional[str] + weight: Optional[float] + wasteType: Optional[str] + bulkIdentity: Optional[str] + + +class Waste(TypedDict, total=False): + """Instance of a manifest waste line""" + + lineNumber: Required[int] + dotHazardous: Required[bool] + epaWaste: Required[bool] + pcb: Required[bool] + dotInformation: Optional[DotInformation] + wasteDescription: Optional[str] + quantity: Optional[Quantity] + brInfo: Optional[BrInfo] + br: bool + hazardousWaste: Optional[HazardousWaste] + pcbInfos: Optional[List[PcbInfo]] + discrepancyResidueInfo: Optional[DiscrepancyResidueInfo] + managementMethod: Optional[RcraCodeDescription] + additionalInfo: Optional[AdditionalInfo] + + +class __BaseManifest(TypedDict, total=False): + """For Internal Use. The bulk of fields in the manifest except for the 'import' field""" + + createdDate: datetime | str + updatedDate: datetime | str + manifestTrackingNumber: str + status: Status + discrepancy: bool + submissionType: SubmissionType + originType: OriginType + shippedDate: datetime | str + receivedDate: datetime | str + generator: RcraSite + transporters: list[RcraSite] + designatedFacility: RcraSite + additionalInfo: AdditionalInfo + wastes: list[Waste] + rejection: bool + residue: bool + # import: bool # This field is added via the ImportFieldMixin + containsPreviousResidueOrRejection: bool + + +# Since we can't use the 'import' keyword as a field name, we add it via this mixin +ImportFieldMixin = TypedDict("ImportFieldMixin", {"import": bool}) + + +class Manifest(__BaseManifest, ImportFieldMixin): + """The RCRA uniform hazardous waste manifest""" + + pass + + +class ManifestExistsResponse(TypedDict): + """manifest exists service response""" + + manifestTrackingNumber: str + result: bool + originType: OriginType + submissionType: SubmissionType + status: Status