3phi-framework 0.2.9

Framework for 3phi project

3phi Framework

Utility classes for DB access, S3 interactions, and data processing via Controller Classes.
Distributed on PyPi.

Install name: 3phi-framework
Import package: threephi_framework

---

Installation

Install from PyPi

pip install 3phi-framework

Installing a Development Build (from CI)

Development builds are generated for each pull request and attached as workflow artifacts.

Download the artifact

1. Open the pull request on GitHub
2. Go to the Checks tab
3. Open the CI and Release workflow run
4. Download the artifact named dist-pr
5. Extract the archive locally

It will contain files like:

dist/ 3phi_framework-.whl 3phi_framework-.tar.gz

Install the wheel (recommended)

From the extracted directory:

pip install dist/3phi_framework-*.whl

Alternatively, install the source distribution:

pip install dist/3phi_framework-*.tar.gz

Notes: Wheels are preferred and install faster. Make sure you are using Python ≥ 3.12 (project requirement). Dev builds are temporary and may be deleted after 7 days.

Quickstart

The framework is set up for local development as well as for being used in a deployment. To set up your environment for local development, follow these steps:

Set up virtual environment

execute_data_app.sh expects a virtual environment to be set up under [.venv]. See the python docs on how to set it up.

Seed data

Obtain seed data for the database and the object storage and copy it to:

• 3_db_seed.sql: This should be a sql dump/snapshot of a working 3phi Database, PSQL will automatically seed the DB when it is created using docker compose.
• 3phi: This should be a copy of a bucket from a working object storage. It will be mounted in the minio object storage as a bucket.

Spin up DB and Object Storage

Navigate to docker and run

docker compose up -d

This will bring up a local DB and a MinIO Object Storage seeded with the data you provided.

Run a data app locally

Use the utility script execute_data_app.sh and pass the data app name as an argument, e.g.:

./execute_data_app.sh sm_classifier

In case the script is not executable, make it executable:

chmod +x execute_data_app.sh

The script will install the dependencies in requirements.txt in your virtual environment, set up environment variables as they are listed in .env and execute the data app as a python module.

Object Storage Connectors

The framework abstracts object storage behind BaseConnector so data apps are decoupled from the underlying storage backend. Two implementations are provided out of the box.

S3Connector

For AWS S3 or any S3-compatible storage (the default local dev setup uses MinIO).

from threephi_framework import S3Connector

connector = S3Connector(data_dir_path="timeseries/ready")

Environment variable	Required	Description
S3_ENDPOINT_URL	Yes	Full URL of the S3 endpoint, e.g. http://localhost:19000 for MinIO
S3_ACCESS_KEY	Yes	Access key / username
S3_SECRET_KEY	Yes	Secret key / password

The bucket name is fixed to 3phi. All paths are rooted at s3://3phi/<data_dir_path>.

AzureBlobConnector

For Azure Blob Storage. Requires the adlfs package (pip install adlfs).

from threephi_framework import AzureBlobConnector

connector = AzureBlobConnector(data_dir_path="timeseries/ready")

Environment variable	Required	Description
AZURE_STORAGE_ACCOUNT_NAME	Yes	Azure Storage Account name
AZURE_STORAGE_CONTAINER_NAME	Yes	Blob container name (equivalent to the S3 bucket)
AZURE_STORAGE_ACCOUNT_KEY	No	Account key for key-based auth. If omitted, DefaultAzureCredential is used automatically

All paths are rooted at az://<container>/<data_dir_path>.

Authentication — when AZURE_STORAGE_ACCOUNT_KEY is not set, the connector falls back to DefaultAzureCredential, which transparently supports managed identity, service principal (via environment variables), and az login for local development. No code changes are needed between environments.

Writing a custom connector

Subclass BaseConnector and implement all abstract methods. The connector is injected into TimeSeriesController at construction time, so any conforming implementation works as a drop-in replacement:

from threephi_framework.object_storage.base_connector import BaseConnector

class MyConnector(BaseConnector):
    ...

controller = TimeSeriesController(connector=MyConnector(data_dir_path="..."))

---

Data Model

The currently assumed datamodel is illustrated in the diagram below: