Prefect Workflows - Data Harvesters

This repository contains Prefect workflows designed for handling various satellite data processing and API interactions. The workflows are modular, with reusable components and utilities, enabling seamless integration of new workflows and customization of existing ones.

Project structure

./common            # Reusable utilities and shared functionality
    ├── date_formats.py    # Date format utilities
    ├── env.py             # Environment variable helpers
    ├── logger.py          # Logging setup
    ├── pgstac.py          # pgSTAC database operations
    ├── s3.py              # S3 storage operations
    ├── sftp.py            # SFTP utilities
    ├── tasks.py           # Reusable Prefect tasks
./flows             # Prefect workflows
    ├── aldoria         # Aldoria API workflows (KPI, OPM, TDM)
    ├── hemeria           # Hemeria workflows (KPI, OPM, TDM)
    ├── spacetrack      # SpaceTrack workflows (SATCAT database)

Getting Started

Prerequisites

1. Python Version: Ensure you are using Python3.12 or later.
2. Prefect version : Ensure the versions in the requirements and the prefect.yaml file match. (default: 3.4.22)
3. Dependencies: Install the required libraries using pip:

pip install -e .

Configuration

Prefect API

To target your Prefect API instance, configure the project with the following command:

prefect config set PREFECT_API_URL=http://localhost:4200/api

Environment

Depending on the workflow requirements, specific components must be activated via Prefect variables and environment variables. These include configurations for databases, storage, APIs, and file transfer systems.

---

PgSTAC Database

The PgSTAC database stores the STAC objects (collections and items) generated by the workflows.

Prefect Variable	Environment Variable	Description
pg_user	PGUSER	The username for the PostgreSQL database.
pg_host	PGHOST	The host address of the PostgreSQL database.
pg_port	PGPORT	The port number of the PostgreSQL database.
pg_database	PGDATABASE	The name of the PostgreSQL database.
pg-password (secret)	PGPASSWORD	The password for the PostgreSQL database.

---

S3 (Provider)

This S3 instance is used to upload the harvested raw data into a specified bucket. The asset links are included in the STAC objects.

Prefect Variable	Environment Variable	Description
s3_access_key_id	AWS_ACCESS_KEY_ID	The access key for the S3 storage.
s3_region	AWS_DEFAULT_REGION	The region for the S3 storage.
s3_endpoint_url	S3_ENDPOINT_URL	The endpoint URL for the S3 storage.
s3-secret-access-key (secret)	AWS_SECRET_ACCESS_KEY	The secret access key for the S3 storage.
s3_bucket_{provider}	S3_BUCKET	The target bucket for the STAC assets in the S3 storage.

---

SFTP Client

An SFTP connection can be added to workflows for duplicating the data collected by harvesters.

Prefect Variable	Environment Variable	Description
sftp_host	SFTP_HOST	The host address of the SFTP server.
sftp_user	SFTP_USER	The username for the SFTP connection.
sftp_remote_dir	SFTP_REMOTE_DIR	The remote directory for SFTP operations.
sftp_port	SFTP_PORT	The port number for the SFTP server.
sftp-password (secret)	SFTP_PASSWORD	The password for the SFTP connection.

---

Space-Track API

Configure the following variables to enable access to the Space-Track API for the Space-Track workflows.

Prefect Variable	Environment Variable	Description
spacetrack_username	SPACETRACK_USERNAME	The username for the Space-Track API.
spacetrack-password (secret)	SPACETRACK_PASSWORD	The password for the Space-Track API.

---

Aldoria API

To harvest data from the Aldoria provider, configure access to the Aldoria API using these variables.

Prefect Variable	Environment Variable	Description
aldoria_username	ALDORIA_USERNAME	The username for the Aldoria API.
aldoria-password (secret)	ALDORIA_PASSWORD	The password for the Aldoria API.

---

Hemeria Internal S3

This internal S3 instance is specific to Hemeria workflows. It synchronizes with Hemeria's servers to periodically retrieve raw data from the MEDOC station using rsync in a CronJob.

Prefect Variable	Environment Variable	Description
internal_s3_access_key_id	INTERNAL_S3_ACCESS_KEY_ID	The access key for the internal S3 storage.
internal_s3_region	INTERNAL_S3_REGION	The region for the internal S3 storage.
internal_s3_endpoint_url	INTERNAL_S3_ENDPOINT_URL	The endpoint URL for the internal S3 storage.
internal-s3-secret-access-key (secret)	INTERNAL_S3_SECRET_ACCESS_KEY	The secret access key for the internal S3 storage.
internal_s3_bucket	INTERNAL_S3_BUCKET	The bucket name for raw data in the internal S3 storage.

Adding new workflows

1. Create a new subdirectory in ./flows for the workflow category or API provider.
2. Write your workflow in a .py file using Prefect's @flow and @task decorators.
3. Reuse utilities and tasks from ./common where applicable.
4. Test your workflow locally before committing.

Example: Creating a Workflow

The typical harvester workflow involves:

1. Loading configuration from Prefect variables and secrets
2. Initializing connections necessary to the workflow (database, API, S3, SFTP...)
3. Harvesting raw data from the provider
4. Reformating the data as needed (e.g. STAC)
5. Cataloging STAC objects (in database) and publishing the results (S3, SFTP)

from prefect import flow

from common.env import load_common_env
from common.tasks import init_db_connection, init_s3_client, init_sftp_connection


@task
def upload_assets_s3(
    client: S3Client, bucket: Bucket, raw_data):
    ...

@task
def upload_assets_sftp(sftp: Connection, raw_data):
    ...


@task
def upload_items_db(db: PgstacDB, stac_data):
    ...

@flow
async def main(
):
    """ Example Flow."""

    load_common_env(provider="my_provider", db=True, sftp=True)

    s3_client, bucket = await init_s3_client()
    db = await init_db_connection()
    sftp = await init_sftp_connection()

    raw_data, stac_data = harvester()

    upload_assets_s3(s3_client, bucket, raw_data)
    upload_items_db(db, stac_data)
    upload_assets_sftp(sftp, raw_data)

Once you finished developing your workflow, you can deploy it using the Prefect API and schedule a flow run.

Example: running a workflow

1. Create a Prefect deployment for your flow (by editing the ./prefect.yaml file)

deployments:
- name: your_workflow
  entrypoint: flows/your_workflow/flow.py:main
  work_pool: *your_workpool
  schedules:
   - *your_schedule

1. Deploy the flow

prefect deploy -n your_workflow

Note : In order to trigger a flow run, you can do it from the Prefect UI or using the CLI.

prefect deployment run 'main/your_workflow'