azure-genome 0.1.3

A Python package for Azure Genome.

azure-genome

A Python client library for the Azure Genome supply-chain service. Provides authentication, data ingestion, and graph query operations through a single root client.

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,
)

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 four sub-clients as attributes:

Attribute	Purpose
client.data	Upload entity and relationship CSV/GZ files
client.query	Read entities, products, and traverse the supply chain graph
client.control	Start and inspect background jobs
client.deployment	Manage workspaces (preview)

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)

Supported data_type values:

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

4. Query the supply chain graph

Get a single entity or product:

entity = client.query.get_entity("3fa85f64-5717-4562-b3fc-2c963f66afa6")
print(entity.name)

product = client.query.get_product("3fa85f64-5717-4562-b3fc-2c963f66afa6")
print(product.name)

List entities with optional filters:

records = client.query.list_entities(filters={"source": "customer"})
for entity in records.items:
    print(entity.id, entity.name)

Traverse the supply chain graph:

graph = client.query.traverse_graph(
    start_entity_key="3fa85f64-5717-4562-b3fc-2c963f66afa6",
    upstream_depth=2,
    downstream_depth=3,
)

N-tier search:

results = client.query.search_n_tier(
    entity_key="3fa85f64-5717-4562-b3fc-2c963f66afa6"
)

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
DataUploadError	A data upload operation fails
QueryError	A query argument is invalid or the resource is not found

from azure_genome import GenomeError
from azure_genome.utils.exceptions import QueryError

try:
    entity = client.query.get_entity(entity_key)
except QueryError as exc:
    print(f"Query failed: {exc}")
except GenomeError as exc:
    print(f"Service error: {exc}")

License

See LICENSE for details.