leaf_engine.io.api

Submodules

• leaf_engine.io.api.base_caller
• leaf_engine.io.api.caller
• leaf_engine.io.api.crud_caller
• leaf_engine.io.api.crud_openapi
• leaf_engine.io.api.openapi
• leaf_engine.io.api.planning_caller
• leaf_engine.io.api.planning_openapi

Exceptions

LeafAPIException

Common base class for all non-exit exceptions.

Classes

LeafAPICaller

Makes calls to the analytics API. The API documentation is available at:

Functions

validate_lane(record)
validate_location(record)
validate_shipment(record)

Package Contents

exception leaf_engine.io.api.LeafAPIException

Bases: Exception

Common base class for all non-exit exceptions.

class leaf_engine.io.api.LeafAPICaller(endpoint: str | None = API_ENDPOINT, key: str | None = API_KEY)

Makes calls to the analytics API. The API documentation is available at: https://api.leaflogistics.com/analytics/docs.

NOTE that using instances of this class in a Jupyter notebook might cause unexpected behavior - always initialize the LeafAPICaller instance in the same notebook cell where you call its methods.

Parameters:

• endpoint (Optional[str]) –

• key (Optional[str]) –

_summarize_responses(path: str, responses: List[LeafAPIResponse]) → None

Parameters:

• path (str) –

• responses (List[LeafAPIResponse]) –

Return type:

None

check_lane_ids(uuids: List[str], batch_size: int = 1000, verbose: int = 0) → List[bool]

Receives a list of UUIDs and returns an equivalent list of bools indicating if the lane UUID already exists in the database.

Parameters:

• uuids (List[str]) –

• batch_size (int) –

• verbose (int) –

Return type:

List[bool]

check_location_ids(uuids: List[str], batch_size: int = 1000, verbose: int = 0) → List[bool]

Receives a list of UUIDs and returns an equivalent list of bools indicating if the location UUID already exists in the database.

Parameters:

• uuids (List[str]) –

• batch_size (int) –

• verbose (int) –

Return type:

List[bool]

check_raw_shipment_ids(uuids: List[str], batch_size: int = 1000, verbose: int = 0) → List[bool]

Receives a list of UUIDs and returns an equivalent list of bools indicating if the raw shipment UUID already exists in the database.

Parameters:

• uuids (List[str]) –

• batch_size (int) –

• verbose (int) –

Return type:

List[bool]

check_shipment_ids(uuids: List[str], batch_size: int = 1000, verbose: int = 0) → List[bool]

Receives a list of UUIDs and returns an equivalent list of bools indicating if the shipment UUID already exists in the database.

Parameters:

• uuids (List[str]) –

• batch_size (int) –

• verbose (int) –

Return type:

List[bool]

create_batch(batch: dict) → dict

Parameters:

batch (dict) –

Return type:

dict

create_cluster_schema(schema: dict) → dict

Parameters:

schema (dict) –

Return type:

dict

create_dataset(dataset: dict) → dict

Parameters:

dataset (dict) –

Return type:

dict

delete_batches(batch_ids: List[str], cascade_delete: bool = False) → List[LeafAPIResponse]

Deletes batches based on batch id.

If cascade_delete is False, only the batch is deleted from the database. If True, shipments, lanes, locations, and datasets that have the same batch date and company id will be deleted.

Parameters:

• batch_ids (List[str]) –

• cascade_delete (bool) –

Return type:

List[LeafAPIResponse]

delete_datasets(dataset_ids: List[str]) → List[LeafAPIResponse]

Deletes datasets based on dataset id.

Parameters:

dataset_ids (List[str]) –

Return type:

List[LeafAPIResponse]

delete_many(records: List[Any], path: str, n_jobs: int = -1, batch_size: int = -1, verbose: int = 0, **kwargs) → List[LeafAPIResponse]

Makes multiple DELETE requests to the analytics API endpoint defined by path.

Splits up records into individual lists based on batch_size. If batch_size is less than or equal to 0, it will send all records in a single DELETE request.

If multiple DELETE calls are made, they are made in parallel using joblib.

Parameters:

• records (List[Any]) –

• path (str) –

• n_jobs (int) –

• batch_size (int) –

• verbose (int) –

Return type:

List[LeafAPIResponse]

delete_one(path: str, data: List | dict, **kwargs) → LeafAPIResponse

Makes a single DELETE request to the analytics API endpoint defined by path.

Note that the current API endpoints support batching, so in the DELETE call below the json data is passed to the API as a list. If a single dictionary is passed, it is wrapped in a list.

Parameters:

• path (str) – The API path to call.

• data (Union[List, Dict]) – The data to DELETE, either a dict or list of dicts.

Returns:

A list of API responses.

Return type:

List[LeafAPIResponse]

get(path: str, **kwargs) → requests.Response

Parameters:

path (str) –

Return type:

requests.Response

get_batches(batch_id: str | None = None, batch_date: str | None = None, company_id: int | None = None) → pandas.DataFrame

Retrieves batches based on given query parameters.

Any kwarg that is not None is used to filter database batches. The batches returned will meet every filter.

The date should be in “YYYY-MM-DD” format.

Parameters:

• batch_id (Optional[str]) –

• batch_date (Optional[str]) –

• company_id (Optional[int]) –

Return type:

pandas.DataFrame

get_cluster_location(company_id: int, zip: str | None = None, city: str | None = None, state: str | None = None, country: str | None = None) → dict | None

Retrieves the most recent cluster for a given location and company.

company_id is a required argument. Along with company_id either zip or city, state, and country must be passed. If all are passed, zip will take priority.

Parameters:

• company_id (int) –

• zip (Optional[str]) –

• city (Optional[str]) –

• state (Optional[str]) –

• country (Optional[str]) –

Return type:

Optional[dict]

get_cluster_locations(company_id: int) → pandas.DataFrame

Retrieves all cluster locations for a given company.

Parameters:

company_id (int) –

Return type:

pandas.DataFrame

get_cluster_schema(company_id: int, name: str) → dict

Returns the most recent cluster schema with name for company_id.

Parameters:

• company_id (int) – Company ID.

• name (str) – Cluster schema name.

Returns:

Cluster schema dictionary.

Return type:

(dict)

get_cluster_schema_by_id(id: str) → dict | None

Returns cluster schema with id.

Parameters:

id (str) – Cluster schema id.

Returns:

Cluster schema dictionary.

Return type:

(dict)

get_datasets(dataset_id: str | None = None, company_id: str | None = None, url: str | None = None, label: str | None = None, mapper: str | None = None, hash: int | None = None) → pandas.DataFrame

Retrieves datasets based on given query parameters.

Any kwarg that is not None is used to filter database datasets. The datasets returned will meet every filter.

Parameters:

• dataset_id (Optional[str]) –

• company_id (Optional[str]) –

• url (Optional[str]) –

• label (Optional[str]) –

• mapper (Optional[str]) –

• hash (Optional[int]) –

Return type:

pandas.DataFrame

get_lanes(company_id: int) → pandas.DataFrame

Retrieves all lanes for a given company.

Parameters:

company_id (int) –

Return type:

pandas.DataFrame

get_locations(company_id: int) → pandas.DataFrame

Retrieves all locations for a given company.

Parameters:

company_id (int) –

Return type:

pandas.DataFrame

get_point_location(company_id: int, zip: str | None = None, city: str | None = None, state: str | None = None, country: str | None = None) → dict | None

Retrieves the most recent point for a given location and company.

company_id is a required argument. Along with company_id either zip or city, state, and country must be passed. If all are passed, zip will take priority.

Parameters:

• company_id (int) –

• zip (Optional[str]) –

• city (Optional[str]) –

• state (Optional[str]) –

• country (Optional[str]) –

Return type:

Optional[dict]

get_point_locations(company_id: int, exclude_record_type: str = 'RFP') → pandas.DataFrame

Retrieves all point locations for a given company.

Optionally excludes locations with a given record type. Defaults to excluding RFPs.

Parameters:

• company_id (int) –

• exclude_record_type (str) –

Return type:

pandas.DataFrame

get_power_lanes(company_id: int) → pandas.DataFrame

Retrieves all power lanes for a given company.

Parameters:

company_id (int) –

Return type:

pandas.DataFrame

get_ptp_lanes(company_id: int) → pandas.DataFrame

Retrieves all ptp lanes for a given company.

Parameters:

company_id (int) –

Return type:

pandas.DataFrame

get_raw_shipments(batch_date: str | None = None, company_id: int | None = None) → pandas.DataFrame

Retrieves raw shipments based on given query parameters.

Any kwarg that is not None is used to filter database shipments. The shipments returned will meet every filter.

The date should be in “YYYY-MM-DD” format.

Parameters:

• batch_date (Optional[str]) –

• company_id (Optional[int]) –

Return type:

pandas.DataFrame

get_records(path: str, **kwargs) → pandas.DataFrame

Makes multiple GET requests to the analytics API endpoint defined by path to get all pages / all records .

Parameters:

path (str) – The API path to call.

Returns:

A DataFrame with all records (useful in getting all lane and location records for a company_id, see methods below).

Return type:

pd.DataFrame

get_shipments(company_id: int) → pandas.DataFrame

Retrieves all shipments for a given company.

Parameters:

company_id (int) –

Return type:

pandas.DataFrame

insert_lanes(records: List[dict], batch_size: int = 50, verbose: int = 0) → List[LeafAPIResponse]

Parameters:

• records (List[dict]) –

• batch_size (int) –

• verbose (int) –

Return type:

List[LeafAPIResponse]

insert_locations(records: List[dict], batch_size: int = 50, verbose: int = 0) → List[LeafAPIResponse]

Parameters:

• records (List[dict]) –

• batch_size (int) –

• verbose (int) –

Return type:

List[LeafAPIResponse]

insert_raw_shipments(records: List[dict], batch_size: int = 50, verbose: int = 0) → List[LeafAPIResponse]

Parameters:

• records (List[dict]) –

• batch_size (int) –

• verbose (int) –

Return type:

List[LeafAPIResponse]

insert_records(path: str, records: List[dict], batch_size: int = 50, verbose: int = 0) → List[LeafAPIResponse]

Parameters:

• path (str) –

• records (List[dict]) –

• batch_size (int) –

• verbose (int) –

Return type:

List[LeafAPIResponse]

insert_shipments(records: List[dict], batch_size: int = 50, verbose: int = 0) → List[LeafAPIResponse]

Parameters:

• records (List[dict]) –

• batch_size (int) –

• verbose (int) –

Return type:

List[LeafAPIResponse]

post(path: str, data: dict | List) → LeafAPIResponse

Parameters:

• path (str) –

• data (Union[dict, List]) –

Return type:

LeafAPIResponse

post_many(records: List[Any], path: str, n_jobs: int = -1, batch_size: int = -1, verbose: int = 0) → List[LeafAPIResponse]

Makes multiple POST requests to the analytics API endpoint defined by path.

Splits up records into individual lists based on batch_size. If batch_size is less than or equal to 0, it will send all records in a single POST request.

If multiple POST calls are made, they are made in parallel using joblib.

Parameters:

• records (List[Any]) –

• path (str) –

• n_jobs (int) –

• batch_size (int) –

• verbose (int) –

Return type:

List[LeafAPIResponse]

put(path: str, data: dict | List) → LeafAPIResponse

Parameters:

• path (str) –

• data (Union[dict, List]) –

Return type:

LeafAPIResponse

update_cluster_schema(schema: dict) → dict

Parameters:

schema (dict) –

Return type:

dict

endpoint

key

retries

session

leaf_engine.io.api.validate_lane(record)

leaf_engine.io.api.validate_location(record)

leaf_engine.io.api.validate_shipment(record)