Skip to content

T3 API : Python Utils

Jump to bottom
Matt Frisbie edited this page Sep 29, 2025 · 2 revisions

T3 API Python Utils

This is the documentation for t3api-utils Python package.

Authentication helpers

get_authenticated_client_or_error_async

async def get_authenticated_client_or_error_async() -> T3APIClient

Interactive, async entrypoint to obtain an authenticated T3APIClient. Presents a picker for credentials, JWT, or API key auth and returns a ready client.

Returns

  • T3APIClient: an authenticated client.

Raises

  • AuthenticationError on authentication failure.
  • typer.Exit if the user cancels or input is invalid.

Notes

  • JWT/API-key flows are handled under the hood. JWT path runs sync creation but returns the client to async code.

get_authenticated_client_or_error

def get_authenticated_client_or_error(*, auth_method: Optional[str] = None) -> T3APIClient

Interactive, sync wrapper to obtain an authenticated client. If auth_method is omitted, shows a picker (credentials | jwt | api_key).

Params

  • auth_method: optional fixed choice to skip the picker.

Returns

  • T3APIClient.

Raises

  • AuthenticationError, typer.Exit.

Example

client = get_authenticated_client_or_error(auth_method="api_key")

get_jwt_authenticated_client_or_error

def get_jwt_authenticated_client_or_error(*, jwt_token: str) -> T3APIClient

Creates a client using a pre-existing JWT. Does not call /whoami automatically.

Params

  • jwt_token: JWT access token.

Returns

  • T3APIClient.

Raises

  • AuthenticationError (invalid token/creation error).
  • ValueError propagated as AuthenticationError.

When to use

  • You already validated the JWT elsewhere or don’t need a live validation call.

get_jwt_authenticated_client_or_error_with_validation

def get_jwt_authenticated_client_or_error_with_validation(*, jwt_token: str) -> T3APIClient

Same as above but validates the token by calling /v2/auth/whoami. Closes the client and surfaces actionable messages if invalid/expired/forbidden.

Params

  • jwt_token: JWT access token.

Returns

  • T3APIClient (validated).

Raises

  • AuthenticationError with clear reason (invalid/expired/forbidden).
  • ValueError wrapped as AuthenticationError.

get_api_key_authenticated_client_or_error

def get_api_key_authenticated_client_or_error(*, api_key: str, state_code: str) -> T3APIClient

Authenticates using API key + state.

Params

  • api_key: T3 API key.
  • state_code: two-letter state code (e.g., CA, MO, CO, MI).

Returns

  • T3APIClient.

Raises

  • AuthenticationError, ValueError wrapped as AuthenticationError.

License utilities

pick_license

def pick_license(*, api_client: T3APIClient) -> LicenseData

Interactive license picker. Fetches /v2/licenses, shows a table, and returns the selected license.

Params

  • api_client: authenticated client.

Returns

  • LicenseData: selected license dict.

Raises

  • typer.Exit if none found or selection invalid.

Collection loading & persistence

load_collection

def load_collection(
    method: Callable[P, MetrcCollectionResponse],
    max_workers: int | None = None,
    *args: P.args,
    **kwargs: P.kwargs,
) -> List[MetrcObject]

Loads all pages of a paginated collection in parallel and returns a flattened list (data items only).

Params

  • method: function that fetches one page and returns a MetrcCollectionResponse.
  • max_workers: optional thread pool size.
  • *args, **kwargs: forwarded to method.

Returns

  • List[MetrcObject]: all items across pages.

Example

items = load_collection(api.list_packages, license_number="CUL000001")

save_collection_to_json

def save_collection_to_json(
    *,
    objects: List[Dict[str, Any]],
    output_dir: str = "output",
    open_after: bool = False,
    filename_override: Optional[str] = None,
) -> Path

Serializes a list of dicts to JSON in output/ (or custom dir). The default filename uses {index or 'collection'}__{licenseNumber}__{timestamp}.json.

Params

  • objects: non-empty list of records.
  • output_dir: directory to write to.
  • open_after: if True, opens the file (platform-dependent).
  • filename_override: base name override.

Returns

  • Path to the saved file.

Raises

  • ValueError if objects is empty.

save_collection_to_csv

def save_collection_to_csv(
    *,
    objects: List[Dict[str, Any]],
    output_dir: str = "output",
    open_after: bool = False,
    filename_override: Optional[str] = None,
    strip_empty_columns: bool = False,
) -> Path

Serializes records to CSV with smart field ordering. Supports optional empty-column stripping.

Params

  • Same as JSON variant, plus:
  • strip_empty_columns: drop columns that are empty across all rows.

Returns

  • Path to the saved file.

Raises

  • ValueError if objects is empty.

interactive_collection_handler

def interactive_collection_handler(*, data: List[Dict[str, Any]]) -> None

Menu-driven TUI workflow for a loaded collection:

  • Inspect items
  • Filter by CSV matches
  • Save to CSV/JSON (custom paths supported)
  • Load into DuckDB
  • Export DB schema

Params

  • data: list of Metrc-like dicts.

Side effects

  • Prompts via typer, prints via rich.
  • May open saved files.
  • Creates/uses a DuckDB connection for the session.

Notes

  • Tracks state (saved paths, DB loaded) and updates the status banner.

DuckDB / data inspection

load_db

def load_db(*, con: duckdb.DuckDBPyConnection, data: List[Dict[str, Any]]) -> None

Loads nested dictionaries into DuckDB, creating:

  • A root table from flattened top-level records.
  • Child tables for nested objects/arrays, named by each nested object’s data_model.
  • Deduplicates by (implicit) IDs within extracted tables.

Params

  • con: active DuckDB connection.
  • data: structured list of records.

Raises

  • ValueError on malformed inputs/table creation issues.

Example

con = duckdb.connect()
load_db(con=con, data=records)

inspect_collection

def inspect_collection(*, data: Sequence[Dict[str, Any]]) -> None

Launches a Textual TUI inspector:

  • Scrollable JSON with highlighting
  • Search with live filtering
  • Mouse/keyboard navigation

Params

  • data: list/sequence of dicts.

Notes

  • Extracts a friendly collection name for the UI header.

File utilities

pick_file

def pick_file(
    *,
    search_directory: str = ".",
    file_extensions: List[str] = [".csv", ".json", ".txt", ".tsv", ".jsonl"],
    include_subdirectories: bool = False,
    allow_custom_path: bool = True,
    load_content: bool = False,
) -> Union[Path, Dict[str, Any]]

Interactive file picker that lists recent files (size, modified time, type) and optionally loads content.

Params

  • search_directory: where to look.
  • file_extensions: allowed extensions.
  • include_subdirectories: recursive search.
  • allow_custom_path: add “Enter custom path…” option.
  • load_content: if True, returns parsed content.

Returns

  • If load_content=False: Path.
  • If True: {"path": Path, "content": Any, "format": str, "size": int}.

Raises

  • typer.Exit on cancellation/no files (when custom path disallowed).
  • FileNotFoundError if custom path is missing.
  • ValueError for unreadable/invalid content.

Supported parsing

  • .json → json.load
  • .jsonl/.ndjson → list of JSON objects
  • .csv/.tsv → list of dict rows
  • others → raw text

Collection filtering & metadata

match_collection_from_csv

def match_collection_from_csv(
    *,
    data: List[Dict[str, Any]],
    on_no_match: Literal["error", "warn", "skip"] = "warn",
) -> List[Dict[str, Any]]

Filters a collection by exact matching against a user-picked CSV/TSV file. CSV headers must exactly equal collection field names.

Params

  • data: collection to filter.

  • on_no_match: behavior for unmatched rows:

    • "error" → raise on first miss
    • "warn" → log a warning (default)
    • "skip" → silent skip

Returns

  • Matched subset (deduplicated, order-preserving). May be empty.

Raises

  • ValueError if CSV columns don’t exist in the collection, bad format, or empty.
  • typer.Exit if selection is canceled.

CSV example

id,name,status
12345,ProductA,Active
67890,ProductB,Inactive

extract_collection_metadata

def extract_collection_metadata(*, data: Sequence[Dict[str, Any]]) -> tuple[str, str]

Derives two labels from a collection:

  • Collection name: "dataModel__index" if index is present; otherwise "dataModel". Returns "mixed_datamodels" if multiple values exist; "empty_collection" if empty.
  • License number: from licenseNumber. Returns "mixed_licenses" if multiple values exist; "no_license" if empty input.

Params

  • data: sequence of dicts.

Returns

  • (collection_name, license_number).

Examples

extract_collection_metadata(data=[
  {"dataModel": "PACKAGE", "index": "active", "licenseNumber": "CUL00001"},
  {"dataModel": "PACKAGE", "index": "active", "licenseNumber": "CUL00001"},
])
# -> ("PACKAGE__active", "CUL00001")

extract_collection_metadata(data=[
  {"dataModel": "PACKAGE", "licenseNumber": "CUL00001"},
  {"dataModel": "PLANT",   "licenseNumber": "CUL00002"},
])
# -> ("mixed_datamodels", "mixed_licenses")

Usage tips

  • Many functions are interactive and designed for CLIs built on typer + rich.
  • When scripting, prefer the non-interactive constructors in your own code paths (e.g., pass auth_method or use the explicit JWT/API key helpers).
  • For large collections, load_collection parallelizes pagination automatically; adjust max_workers for your environment.
  • Use load_db → duckdb for quick local analytics and export_duckdb_schema (via the interactive handler) to understand the generated schema.

Next Steps

Created by Matt Frisbie

Contact: matt@trackandtracetools

Copyright © 2025 Track & Trace Tools. All rights reserved.

Sidebar

Clone this wiki locally