This post contains a collection of code snippets and examples that demonstrate how to use the Ephany Framework API. As the project evolves, I’ll continue adding practical, copy-and-paste samples that show how to connect to your asset data, query your catalogs, integrate with your applications, and build on top of the framework.

Whether you’re a developer, BIM manager, or someone exploring what an open-source asset management platform can do, consider this your living reference for getting up and running with Ephany.

Assets

Two basic functions demonstrating how to get Assets (and their respective data). These are demonstrated in the example file: /support/examples/assets_get.py.

Get all assets (with pagination):

import os
import requests
from dotenv import load_dotenv

"""
Simple pagination test for an Ephany API endpoint.

- Confirms the response is paginated
- Iterates through all pages using `next`
- Prints how many records are returned per page
"""

load_dotenv()
api_key = os.getenv("EPHANY_API_KEY_DEV")

if not api_key:
    raise RuntimeError("EPHANY_API_KEY environment variable is not set")

headers = {
    "Accept": "application/json",
    "X-API-KEY": api_key,
}

# Request page 2, 20 items per page
url = "http://localhost:8000/api/assets/"
params = {
    "page": 2,
    "page_size": 20,
}

response = requests.get(url, headers=headers, params=params, timeout=30)
response.raise_for_status()
data = response.json()

results = data["results"]

print(f"Returned {len(results)} assets")
print(f"Total assets available: {data['count']}")

Search assets for a specific keyword:

import os
import math
import requests
from dotenv import load_dotenv

"""
Pagination + keyword search example for the Ephany API.

- Sends a keyword search query (?search=...)
- Requests a specific page and page size
- Prints "Page X of Y" plus basic pagination metadata
- Prints the model of each returned asset
"""

load_dotenv()

API_KEY_ENV = "YOUR EPHANY API KEY"
api_key = os.getenv(API_KEY_ENV)
if not api_key:
    raise RuntimeError(f"{API_KEY_ENV} environment variable is not set")

headers = {
    "Accept": "application/json",
    "X-API-KEY": api_key,
}

url = "http://localhost:8000/api/assets/"

params = {
    "search": "Avantco",
    "page": 4,
    "page_size": 10,
}

response = requests.get(url, headers=headers, params=params, timeout=30)
response.raise_for_status()
data = response.json()

results = data.get("results", [])
count = int(data.get("count", 0))

page = int(params.get("page", 1))
page_size = int(params.get("page_size", 10))
total_pages = max(1, math.ceil(count / page_size)) if page_size > 0 else 1

# Print models returned on this page
# for asset in results:
#     print(asset.get("model", ""))

print(f"\nSearch term: {params['search']}")
print(f"Page {page} of {total_pages}")
print(f"Returned {len(results)} assets on this page")
print(f"Total matching assets: {count}")

Update a Property Value

import os
import requests
from dotenv import load_dotenv

"""
Bulk update example for the Ephany API.

Goal:
- Find all assets where manufacturer_name == "Avantco"
- Update the asset URL to:
  https://www.avantcorefrigeration.com/product/{asset.sku}

Notes:
- Uses API key authentication via environment variable
- Assumes `sku` already exists (either as a top-level field or in custom_fields)
- Handles paginated API responses
"""

load_dotenv()

API_KEY_ENV = "EPHANY_API_KEY_DEV"
api_key = os.getenv(API_KEY_ENV)
if not api_key:
    raise RuntimeError(f"{API_KEY_ENV} environment variable is not set")

BASE_URL = "http://localhost:8000/api/assets/"

headers = {
    "Accept": "application/json",
    "Content-Type": "application/json",
    "X-API-KEY": api_key,
}


def build_product_url(sku: str) -> str:
    """Construct the Avantco product URL from a SKU."""
    return f"https://www.avantcorefrigeration.com/product/{sku}"


def update_assets_for_manufacturer(manufacturer_name: str):
    url = BASE_URL
    params = {
        "manufacturer__name": manufacturer_name,
        "page_size": 50,
    }

    updated = 0
    scanned = 0

    while url:
        response = requests.get(url, headers=headers, params=params, timeout=30)
        response.raise_for_status()
        data = response.json()

        results = data.get("results", [])
        scanned += len(results)

        for asset in results:
            asset_id = asset.get("id")

            # Pull SKU from either top-level field or custom_fields
            sku = asset.get("sku")
            if not sku:
                custom_fields = asset.get("custom_fields") or {}
                sku = custom_fields.get("sku")

            if not sku:
                print(f"[SKIP] Asset {asset_id}: no SKU found")
                continue

            new_url = build_product_url(sku)

            patch_resp = requests.patch(
                f"{BASE_URL}{asset_id}/",
                headers=headers,
                json={"url": new_url},
                timeout=30,
            )

            if patch_resp.status_code == 200:
                print(f"[OK] Asset {asset_id} → {new_url}")
                updated += 1
            else:
                print(
                    f"[FAIL] Asset {asset_id} "
                    f"({patch_resp.status_code}): {patch_resp.text}"
                )

        url = data.get("next")
        params = None  # next already includes query params

    print("\nUpdate complete")
    print(f"Scanned assets: {scanned}")
    print(f"Updated assets: {updated}")


if __name__ == "__main__":
    update_assets_for_manufacturer("Avantco")

Update the Asset Category

def update_asset_category():
    print("Starting update job...")

    next_url = f"{BASE_URL}"
    params = {"manufacturer": 3}

    total_processed = 0

    while next_url:
        try:
            print(f"Fetching page: {next_url}")
            # Params are only needed for the first request; subsequent 'next' links usually include them
            response = requests.get(next_url, params=params if next_url == f"{BASE_URL}" else None,
                                    headers=headers)

            if response.status_code != 200:
                print(f"Critical Error fetching list: {response.status_code} - {response.text}")
                break

            data = response.json()

            # Extract results and the next page link
            if isinstance(data, dict) and 'results' in data:
                assets = data['results']
                next_url = data.get('next')  # Get the URL for the next page, or None if finished
            else:
                # Handle cases where API might not be paginated
                assets = data
                next_url = None

            if not assets:
                print("No assets found on this page.")
                break

            # Process the current page
            for asset in assets:
                total_processed += 1
                asset_id = asset.get('id')
                asset_name = asset.get('name', 'Unknown Asset')

                detail_url = f"{BASE_URL}{asset_id}/"

                payload = {
                    "category_id": 3
                }

                print(f"[{total_processed}] Updating Asset {asset_id} ({asset_name})...", end=" ")

                patch_response = requests.patch(detail_url, json=payload, headers=headers)

                if patch_response.status_code in [200, 204]:
                    print("Success.")
                else:
                    print(f"Failed ({patch_response.status_code}).")
                    print(f"Reason: {patch_response.text}")

                # Sleep briefly to avoid rate limits
                # time.sleep(0.1)

        except requests.exceptions.RequestException as e:
            print(f"Network error occurred: {e}")
            break

    print(f"Job complete. Processed {total_processed} assets.")