azure-genome 0.1.4

A Python package for Azure Genome.

azure-genome

A Python client library for the Azure Genome supply-chain service. The current release focuses on authenticated data ingestion — uploading entity, product, and relationship files and orchestrating graph builds.

Installation

pip install azure-genome

Requires Python 3.11 or later.

Quick start

1. Authenticate

Static bearer token — use when you already have a short-lived token (e.g. from CI or a test harness):

from azure_genome import StaticTokenCredential

credential = StaticTokenCredential(access_token="<your-bearer-token>")

Certificate-based — use GenomeCertificateCredential for production workloads that authenticate via a client certificate registered in Entra ID. The recommended approach is to store the certificate in Azure Key Vault and load it at runtime. Key Vault exposes the certificate bundle (certificate + private key) as a base64-encoded PFX through its Secrets API:

pip install azure-keyvault-secrets azure-identity

import base64
from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
from cryptography.hazmat.primitives.serialization.pkcs12 import load_pkcs12
from azure_genome import GenomeCertificateCredential

# Fetch the certificate bundle from Key Vault
kv_credential = DefaultAzureCredential()
secret_client = SecretClient(
    vault_url="https://<vault-name>.vault.azure.net",
    credential=kv_credential,
)
secret = secret_client.get_secret("<certificate-name>")
pfx_bytes = base64.b64decode(secret.value)

# Parse the PFX to extract the private key and certificate
pkcs12 = load_pkcs12(pfx_bytes, password=None)
private_key = pkcs12.key
certificate = pkcs12.cert.certificate

credential = GenomeCertificateCredential(
    tenant_id="<tenant-id>",
    client_id="<client-id>",
    scope="<scope>",
    private_key=private_key,
    certificate=certificate,
)

The certificate credential caches tokens in-memory and refreshes them automatically ~30 minutes before expiry.

2. Create a client

from azure_genome import GenomeClient

client = GenomeClient(
    endpoint="https://<your-genome-service>.example.com",
    credential=credential,
)

Note: Only HTTPS endpoints are accepted. The client will raise a TransportError if a non-HTTPS endpoint is provided.

The root client exposes the following surface today:

Attribute	Status	Purpose
client.data	Stable	Upload entity, product, and relationship files; trigger and monitor graph builds
client.swagger	Stable	Fetch the live OpenAPI specification from the configured endpoint
client.query	Preview / WIP	Read entities, products, and traverse the supply chain graph
client.control	Preview / WIP	Start and inspect background jobs
client.deployment	Preview / WIP	Manage workspaces

The query, control, and deployment sub-clients are under active development and their surfaces may change. This README only documents the stable data-ingestion flow.

3. Upload data

# Upload a full entity file (CSV or gzip)
response = client.data.upload_data(
    data_type="entity",
    file_path="entities.csv",
    push_type="full",   # or "delta"
)
print(response.requestId, response.status)

Supported data_type values:

• entity, product
• entityaddressrel, entityflagrel, entityidentifierrel
• entityownsentityrel, entitysuppliedbyentityrel
• entitysellsproductrel, entitybuysproductrel
• productcontainscomponentrel, producthasvariantrel, productflagrel

push_type accepts "full" (replace all records of this type) or "delta" (apply incremental changes). Files must end in .csv or .gz.

4. Track upload status

# Poll a single request
status = client.data.check_upload_status(response.requestId)
print(status.status, status.errorMessage)

# Or list every upload submitted to the service
all_uploads = client.data.list_uploads()
for upload in all_uploads.uploads or []:
    print(upload.requestId, upload.dataType, upload.status)

5. Build the graph

After every file in a batch reports "Completed", trigger a graph build so the service incorporates the new data:

import time

build = client.data.build_graph()
print(build.buildId, build.status)

# Poll until terminal
while True:
    result = client.data.check_build_status(build.buildId)
    if result.status in ("Completed", "Failed"):
        break
    time.sleep(5)

print(result.status, result.errorMessage)

List previous builds:

for b in (client.data.list_builds().builds or []):
    print(b.buildId, b.status, b.modifiedOn)

6. Inspect the service contract (optional)

spec = client.swagger   # returns the parsed OpenAPI JSON dict
print(spec["info"]["title"], spec["info"]["version"])

Error handling

All exceptions inherit from GenomeError:

Exception	Raised when
AuthenticationError	Token or certificate details are missing or invalid
TransportError	An HTTP request cannot be prepared or sent (includes non-HTTPS endpoints and non-2xx responses)
DataUploadError	An upload argument is invalid or the upload fails
GenomeError	Base class — catch this to handle any service-side failure

from azure_genome import GenomeError
from azure_genome.utils.exceptions import DataUploadError, TransportError

try:
    response = client.data.upload_data(
        data_type="entity",
        file_path="entities.csv",
    )
except DataUploadError as exc:
    print(f"Upload rejected: {exc}")
except TransportError as exc:
    print(f"Network or HTTP failure: {exc}")
except GenomeError as exc:
    print(f"Service error: {exc}")

Logging

The library logs request lifecycle events to the azure_genome logger hierarchy at INFO. Enable it like any standard logger:

import logging
logging.basicConfig(level=logging.INFO)
logging.getLogger("azure_genome").setLevel(logging.INFO)

Compatibility

• Python: 3.11, 3.12, 3.13, 3.14
• Transport: HTTPS only — bearer tokens are never sent over unencrypted connections

License

See LICENSE for details.