LucoPy 1.2.1

Python SDK to support the Luco data observability tool.

LucoPy

Giles Matthews: [e-mail]

Purpose

A generalised Python interface to interact with the Luco API from any python environment.

---

How to Use

Install the latest version from PyPI using PIP

pip install LucoPy

Or, specify a version

pip install LucoPy==1.2.1

Import the LucoApi class from the LucoPy library in your script and create an object of this class by passing in the API base url and the appropriate credentials. E.g.

from LucoPy import LucoApi

api = LucoApi(base_url, tenant_id, client_id, client_secret, resource_id)

---

Authentication

In order to make calls to the API endpoints, LucoPy must be able to generate an authenticated access token. Authentication is managed by the ApiCore class using an identity provided during the instantiation of the LucoApi object.

Azure Service Principal

If the Luco instance is hosted on Azure, a service principal can be used to authenticate to the API. In order to use a service principal, an App Registration must be created in the same Azure subscription as the Luco instance. The required credentials must then be passed as arguments to the LucoApi class when instantiating it. These credentials are:

• tenant_id - Directory (tenant) ID of the App Registration representing LucoPy
• client_id - Application (client) ID of the App Registration representing LucoPy
• client_secret - Secret value of the App Registration representing LucoPy
• resource_id - Application (client) ID of the target App Registration representing the API

Other Identities

Any custom identity object can be passed into the LucoApi class via the identity kwarg when instantiating the LucoApi class. This idenity object must have a method called generate_token() which returns an access token validated to the API and the expiry datetime of this token.

---

LucoApi Class

LucoApi(base_url, tenant_id=None, client_id=None, client_secret=None, resource_id=None, identity=None, timeout=20, log=False)

This class acts as the gateway to the Luco platform. An instance of this class should be created at the beginning of each script, API calls are then made through the ApiCore which handles the necessary authentication.

The base URL of the API instance must be passed as a parameter to this object along with the method of authentication.

The timeout option defines the maximum time (seconds) to wait for an HTTPS response from the API before causing a failure.

Use the log argument to turn logging on or off. Logs are generated and sent to a log.txt file in the base directory alongside where the script is being run.

1. find_slot_id(tag, slot_sequence)

   Find slot id from a tag/date and slot sequence definition. If the slot sequence does not have an active delivery schedule and a new tag is provided - a new slot will be created with this tag and the id of this slot will be returned.

   Args:

   • tag (str) : Date (YYYY-MM-DD) for scheduled deliveries or Unique tag for unscheduled deliveries
   • slot_sequence (list of k:v pairs (dicts)) : list slot sequence definitions in form {'key': 'value'}. Order matters - this determines parameter position.

   Returns:

   • slot_id (int)

2. get_submission(slot_id, submission_id)

   Returns a submission object representing an existing submission.

   Args:

   • slot_id (int)
   • submission_id (int)

   Returns:

   • submission (Submission)

3. create_submission(slot_id)

   Create a submission against a slot and returns a Submission object representing it.

   Args:

   • slot_id (int)
   • stage (string) : None
   • run_environment (dict or list of dicts) : None

   Returns:

   • submission (Submission)

4. find_submission_in_slot_sequence(slotId, submissionId, OnlyCompletedSubmissions=False, TimeDifference=None, FindClosest='historic')

   Returns a Slot and Submission ID and whether it is an exact match based on the search criteria, and what the relative difference is in terms of time and number of slots.

   Args:

   • slot_id (int)
   • submission_id (int)
   • OnlyCompletedSubmissions (bool)
   • TimeDifference (str) : d:HH:MM:SS
   • FindClosest (str) : historic, future, either or exact

   Returns:

   • Response JSON (dict)

5. find_submissions_by_slot_sequence(slotSequence, onlyLatestSlot=True, onlyDeliveredSlots=True, onlyCompletedSubmissions=True, onlyLatestSubmission=True, expectedAfterUtc=None, expectedBeforeUtc=None)

   Returns submissions and their slots for a slot sequence

   Args:

   • slotSequence (list of k:v pairs (dicts))
   • onlyLatestSlot (bool)
   • onlyDeliveredSlots (bool)
   • onlyCompletedSubmissions (bool)
   • onlyLatestSubmission (bool)
   • expectedAfterUtc (str) : YYYY-MM-DD or YYYY-MM-DDThh:mm:ss
   • expectedBeforeUtc (str) : YYYY-MM-DD or YYYY-MM-DDThh:mm:ss

   Returns:

   • Response JSON (dict)

6. find_latest_submission_by_slot_sequence(slotSequence, expectedAfterUtc=None, expectedBeforeUtc=None)

   Accessory method to find_submissions_by_slot_sequence(). Returns the slot id and submission id of the most recently completed submission on the slot sequence.

   Equivalent to:

   find_submissions_by_slot_sequence(slotSequence, expectedAfterUtc=expectedAfterUtc, expectedBeforeUtc=expectedBeforeUtc)

   Where the response JSON is interpreted to only return the slot id and submission id.

   Args:

   • slotSequence (list of k:v pairs (dicts))
   • expectedAfterUtc (str) : YYYY-MM-DD or YYYY-MM-DDThh:mm:ss
   • expectedBeforeUtc (str) : YYYY-MM-DD or YYYY-MM-DDThh:mm:ss

   Returns:

   • slot_id (int), submission_id (int)

Submission Class

Submission(slot_id, submission_id, core)

Much of the functionality is handled at the Submission level. A Submission object is created by the get_submission or create_submission methods of the LucoApi class. These objects store the definition of the corresponding submission and handle methods relating to it.

1. params(group=None, key=None)

   Retrieve slot parameters. The group and key kwargs can be used to refine the response. Only use key in addition to group.

   Args:

   • group (str) : Parameter group to return
   • key (str) : Key within group to return the value of

   Returns:

   • result (dict or str)

2. get_metrics(stages=None, metrics=None)

   Retrieve metrics. Filter by stage and metric by passing strings or lists of strings.

   Args:

   • stages (string)
   • metrics (string)

   Returns:

   • metrics (dict)

3. get_quality() --> dict

   Retrieve quality results

   Returns:

   • quality (dict)

4. submit_run_environment(stage=None, run_environments=None)

   Submit run environment details

   Args:

   • stage (string) : Optional
   • run_environments (dict or list of dicts) : Required

   Returns:

   • response status (Bool) : Boolen success or failure

5. submit_metrics(stage, metric=None, value=None, metrics=None)

   Submit metrics by passing a dict of metric : value pairs to metrics. Option to pass a single metric : value pair using metric and value. It is recommended to use metrics.

   Args:

   • stage (str)
   • metric (str) : Metric key
   • value (str) : Value of metric
   • metrics (dict) : Dictionary of Metric : Value pairs.

   Returns:

   • response status (Bool) : Boolen success or failure

6. submit_quality(stage, tool=None, results=None, dataset=None, action=None)

   Submit quality results

   Args:

   • stage (str)
   • tool (str)
   • results (str)
   • dataset (str) : Optional
   • action (str)

   Returns:

   • response status (Bool) : Boolen success or failure

7. submit_status(status, stage=None, type=None, message=None, modified_by=None)

   Submit the status of the Submission

   Args:

   • status (str)
   • stage (str) : Optional
   • type (str) : Optional
   • message (str) : Optional
   • modified_by (str) : Optional

   Returns:

   • response status (Bool) : Boolen success or failure

8. submit_completed_status()

   Submit a status for a completed submission. Equivalent to: submit_status('Completed', 'Submission')

   Returns:

   • response status (Bool) : Boolen success or failure

---

Version History

LucoPy-1.2.1 : Updated find_slot_id method to use new POST /slots/ endpoint to create unscheduled slots. No change to user.

LucoPy-1.2.0 : First version hosted on PyPI.

---