Apache Airflow Integration

pip install "semantica[ingest-airflow]"

from semantica.ingest import AirflowIngestor

ingestor = AirflowIngestor(
    base_url="https://airflow.example.com",
    token="your-access-token",
)

data = ingestor.ingest()
documents = ingestor.export_as_documents(data)

print(f"DAGs: {len(data.dags)}")
print(f"Tasks: {len(data.tasks)}")
print(f"Dependencies: {len(data.dependencies)}")

ingestor.close()

from semantica.ingest import AirflowIngestor

ingestor = AirflowIngestor(
    base_url="https://airflow.example.com",
    token="your-access-token",
)

from semantica.ingest import AirflowIngestor

ingestor = AirflowIngestor(
    base_url="https://airflow.example.com",
    username="airflow-user",
    password="your-password",
)

data = ingestor.ingest(
    dag_ids=["daily_etl", "warehouse_sync"],
)

data = ingestor.ingest(
    include_paused=False,
)

data = ingestor.ingest(
    dag_ids=["daily_etl", "warehouse_sync"],
    include_paused=False,
)

[
    {
        "dag_id": "daily_etl",
        "upstream_task_id": "extract",
        "downstream_task_id": "transform",
    },
    {
        "dag_id": "daily_etl",
        "upstream_task_id": "transform",
        "downstream_task_id": "load",
    },
]

airflow:dag:<dag_id>
airflow:task:<dag_id>:<task_id>
airflow:dependency:<dag_id>:<upstream_task_id>:<downstream_task_id>

from semantica.ingest import AirflowIngestor

ingestor = AirflowIngestor(
    base_url="http://10.0.0.20:8080",
    username="airflow-user",
    password="your-password",
    allow_private_ips=True,
)

data = ingestor.ingest(
    page_limit=100,
)

GET /api/v1/dags
GET /api/v1/dags/{dag_id}/tasks

ingestor.close()