from prefect.blocks.system import Secret

# Create and save once (e.g., in a setup script or REPL)
Secret(value="YOUR_API_KEY").save("my-api-key")

# Load and use in your flow
from prefect import flow
from prefect.blocks.system import Secret

@flow
def my_flow():
    api_key = Secret.load("my-api-key").get()
    # use api_key safely here

if __name__ == "__main__":
    my_flow()

task_run

flow_run_id

NOT NULL

prefect server start

uv run prefect config view --show-sources

server.database.connection_url

home

sqlite+aiosqlite:///...

$PREFECT_HOME

server.database.connection_url

uv run prefect server database reset -y

sqlite3 /path/to/your.db "DELETE FROM task_run WHERE flow_run_id IS NULL;"

psql "$PREFECT_SERVER_DATABASE_CONNECTION_URL" -c "DELETE FROM task_run WHERE flow_run_id IS NULL;"

uv run prefect server database reset -y

PREFECT_SERVER_DATABASE_CONNECTION_URL="sqlite+aiosqlite:///absolute/path/to/your.db" uv run prefect server database reset -y

uv run prefect version
uv run prefect config view --show-sources --show-defaults

uv run prefect config view --show-sources --show-secrets | grep DATABASE_CONNECTION_URL

PREFECT_API_DATABASE_CONNECTION_URL='sqlite+aiosqlite:///.../.prefect/prefect.db'

rm -f ~/.prefect/prefect.db

uv run prefect server database reset -y

uv run prefect server start

PREFECT_API_DATABASE_CONNECTION_URL="sqlite+aiosqlite:///$HOME/.prefect/prefect.db" uv run prefect server database reset -y

uv run prefect version

uv run prefect server start

PREFECT_API_URL=<http://127.0.0.1:4200/api>

--no-services

# stop any running server
# then start it cleanly
uv run prefect server start

# stop server first
uv run prefect config view --show-sources --show-secrets | grep DATABASE_CONNECTION_URL
# delete the sqlite file that command shows (typically ~/.prefect/prefect.db)
rm -f ~/.prefect/prefect.db
uv run prefect server database reset -y
uv run prefect server start

.submit()

.result()

from prefect import flow, task

@task
def plus_one(x): 
    return x + 1

@task
def times_two(x): 
    return x * 2

@flow
def demo():
    a = plus_one(1)
    b = times_two(a)
    return b

if __name__ == "__main__":
    demo()

@flow
def demo_concurrent():
    a_f = plus_one.submit(1)
    b_f = times_two.submit(a_f)
    b = b_f.result()  # ensure completion
    return b

fetch_weather

save_weather

prefect server start

uv run prefect config view --show-sources --show-secrets | grep DATABASE_CONNECTION_URL

~/.prefect/prefect.db

rm -f ~/.prefect/prefect.db

uv run prefect server database reset -y
uv run prefect server start

--no-services

export PREFECT_API_URL="<http://127.0.0.1:4200/api>"

from prefect import flow, task
import httpx

@task(log_prints=True)
def fetch_weather(lat: float, lon: float):
    base_url = "<https://api.open-meteo.com/v1/forecast>"
    temps = httpx.get(
        base_url,
        params=dict(latitude=lat, longitude=lon, hourly="temperature_2m"),
    )
    forecasted_temp = float(temps.json()["hourly"]["temperature_2m"][0])
    print(f"Forecasted temp C: {forecasted_temp} degrees")
    return forecasted_temp

@task(log_prints=True)
def save_weather(temp: float):
    print(f"Saving {temp} to weather.csv")
    with open("weather.csv", "w+") as w:
        w.write(str(temp))
    return "Successfully wrote temp"

@flow(log_prints=True)
def pipeline(lat: float = 38.9, lon: float = -77.0):
    temp = fetch_weather(lat, lon)
    result = save_weather(temp)
    return result

if __name__ == "__main__":
    pipeline()

uv run prefect version

uv run prefect server start

PREFECT_API_URL

time.sleep(1)

--no-services

uv run prefect server start

PREFECT_API_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL=1 uv run prefect server start

@task(log_prints=True)
  def save_weather(temp: float):
      print(f"Saving {temp}…")
      ...

uv run prefect version

PREFECT_API_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL=1

PREFECT_API_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL

PREFECT_API_SERVICES_EVENT_PERSISTER_BATCH_SIZE

PREFECT_API_SERVICES_EVENT_PERSISTER_ENABLED

PREFECT_API_EVENTS_STREAM_OUT_ENABLED

PREFECT_API_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL=1 uv run prefect server start

uv run prefect config set PREFECT_API_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL=1
uv run prefect config view --show-sources
# restart the server to pick up the new setting
uv run prefect server start

PREFECT_SERVER_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL

prefect server start

prefect config set ...

$PREFECT_HOME/profiles.toml

~/.prefect/profiles.toml

uv run prefect config view --show-sources
uv run prefect profile ls

uv run prefect config unset PREFECT_API_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL -y
# or set a new value
uv run prefect config set PREFECT_API_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL=5

PREFECT_API_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL

PREFECT_API_SERVICES_EVENT_PERSISTER_BATCH_SIZE

uv run prefect server start

uv run prefect profile use another-profile

PREFECT_SERVER_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL

uv run prefect config unset PREFECT_API_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL -y

uv run prefect server start

uv run prefect config view --show-sources

unset PREFECT_API_SERVICES_EVENT_PERSISTER_FLUSH_INTERVAL

PREFECT_LOGGING_LOG_PRINTS

PREFECT_LOGGING_LOG_PRINTS=true

uv run prefect config set PREFECT_LOGGING_LOG_PRINTS=true

@flow(log_prints=True)

@task(log_prints=True)

log_prints

{
  "lat": [],
  "lon": []
}

{
  "lat": 38.9,
  "lon": -77
}

{
  "temp": [
    {
      "inputType": "task_run",
      "id": "<upstream task run id>"
    }
  ]
}

temp

fetch_weather

{
  "lat": [],
  "lon": []
}

inputType: "parameter"

inputType: "constant"

from prefect import task, flow

  @task(log_prints=True)
  def fetch_weather(lat: float, lon: float):
      print(f"lat={lat}, lon={lon}")
      ...

from prefect import artifacts

  @task
  def save_weather(temp: float):
      artifacts.create_markdown(f"Temp: {temp}")
      ...

uv run prefect block type ls

uv run prefect block create secret

uv run prefect block inspect secret/secret-block

uv run python -c "from prefect.blocks.system import Secret; Secret(value='chave').save('secret-block')"

uv run prefect block ls
uv run prefect block inspect secret/secret-block

from prefect.blocks.system import Secret
api_key = Secret.load("secret-block").get()

@task(persist_result=True)

import asyncio
from prefect.client.orchestration import get_client
from prefect.results import get_state_result

async def load_latest_my_task_result():
    async with get_client() as client:
        # get the most recent run of 'my_task'
        trs = await client.read_task_runs(
            task_run_filter={"name": {"any_": ["my_task"]}},
            limit=1,
        )
        state = trs[0].state
        df = await get_state_result(state)  # fetches from storage and deserializes
        return df

df = asyncio.run(load_latest_my_task_result())
print(df)

import os, cloudpickle

path = os.path.expanduser("~/.prefect/storage/<storage_key>")
with open(path, "rb") as f:
    df = cloudpickle.load(f)

get_state_result

result_storage=...

PREFECT_RESULTS_DEFAULT_SERIALIZER

PREFECT_TASKS_DEFAULT_PERSIST_RESULT=true

from prefect.blocks.notifications import SlackWebhook

  @flow
  def my_flow():
      SlackWebhook.load("my-slack").notify("Flow finished!")

from my_flow_module import my_flow
my_flow()                 # ad-hoc run
# or
from prefect import flow
@flow
def my_flow(...):
    ...
my_flow.serve(name="local")   # local scheduler, no deployments needed

from prefect import flow
from prefect.storage import GitRepository

# Option A: direct URL string
flow.from_source(
    source="<https://github.com/owner/repo>",
    entrypoint="path/to/flow_file.py:my_flow",
).deploy(
    name="my-deployment",
    work_pool_name="process-pool"   # local process worker
)

# Option B: explicit GitRepository (choose branch/commit)
repo = GitRepository(
    url="<https://github.com/owner/repo>",
    branch="main",   # or commit_sha="abc123"
)
flow.from_source(repo, "path/to/flow_file.py:my_flow").deploy(
    name="my-deployment",
    work_pool_name="process-pool"
)

from prefect.storage import GitRepository, GitCredentials
from prefect.blocks.system import Secret

# Save your token once
Secret(value="ghp_xxx").save("github-token")

creds = GitCredentials(access_token=Secret.load("github-token").get())
repo = GitRepository(url="<https://github.com/owner/private-repo>", credentials=creds)

flow.from_source(repo, "flows/my_flow.py:my_flow").deploy(
    name="secure-deployment",
    work_pool_name="process-pool"
)

uv run prefect work-pool create process-pool --type process
uv run prefect worker start -p process-pool

uv run prefect deployment run "my-flow/my-deployment"

Deployment.build_from_flow()

prefect deployment build

flow.serve()

uv run prefect config view --show-sources

PREFECT_API_URL

uv run prefect deployment ls
uv run prefect deployment inspect "pipeline/my-first-managed-deployment"

from prefect import flow

@flow
def pipeline(lat: float = 38.9, lon: float = -77.0):
    ...

# run once to create the deployment
pipeline.deploy(
    name="my-first-managed-deployment",
    work_pool_name="process-pool",
)

uv run prefect deploy ./path/to/file.py:pipeline --name my-first-managed-deployment

uv run prefect work-pool create process-pool --type process
uv run prefect worker start -p process-pool

from prefect import flow
from prefect.deployments import run_deployment

@flow
def run_deployment_from_flow():
    run_deployment(
        name="pipeline/my-first-managed-deployment",
        parameters={"lat": 1, "lon": 2},
    )

uv run prefect deployment ls

uv run prefect config view --show-sources

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
import httpx

@task(retries=2, retry_delay_seconds=5, timeout_seconds=60)
def fetch_api1():
    return httpx.get("<https://api.example.com/a>").json()

@task(retries=2, retry_delay_seconds=5, timeout_seconds=60)
def fetch_api2():
    return httpx.get("<https://api.example.com/b>").json()

@task(retries=2, retry_delay_seconds=5)
def download_file(url: str) -> str:
    # stream to disk; don’t return large bytes
    path = f"data/{url.split('/')[-1]}"
    with httpx.stream("GET", url) as r:
        r.raise_for_status()
        with open(path, "wb") as f:
            for chunk in r.iter_bytes():
                f.write(chunk)
    return path  # return the path, not file contents

@flow(task_runner=ThreadPoolTaskRunner(max_workers=8))
def etl(urls: list[str]):
    a1_f = fetch_api1.submit()
    a2_f = fetch_api2.submit()
    file_fs = [download_file.submit(u) for u in urls]

    a1 = a1_f.result()
    a2 = a2_f.result()
    paths = [f.result() for f in file_fs]
    return a1, a2, paths

if __name__ == "__main__":
    etl(["<https://example.com/large1.bin>", "<https://example.com/large2.bin>"])

from prefect import flow
from prefect.deployments import run_deployment

@flow
def orchestrator():
    # kick off two independent deployed flows in parallel
    fr1 = run_deployment(name="api-a-flow/prod")
    fr2 = run_deployment(name="api-b-flow/prod")
    # optionally wait / poll if you need their results

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
import httpx

@task(retries=2, retry_delay_seconds=5, timeout_seconds=60)
def api_a():
    return httpx.get("<https://api.example.com/a>").json()

@task(retries=2, retry_delay_seconds=5, timeout_seconds=60)
def api_b():
    return httpx.get("<https://api.example.com/b>").json()

@task(retries=2, retry_delay_seconds=5)
def big_file(url: str) -> str:
    path = f"data/{url.split('/')[-1]}"
    with httpx.stream("GET", url) as r:
        r.raise_for_status()
        with open(path, "wb") as f:
            for chunk in r.iter_bytes():
                f.write(chunk)
    return path

@flow(task_runner=ThreadPoolTaskRunner(max_workers=8))
def pipeline(urls: list[str]):
    a_f = api_a.submit()
    b_f = api_b.submit()
    file_fs = [big_file.submit(u) for u in urls]

    a = a_f.result()
    b = b_f.result()
    paths = [f.result() for f in file_fs]
    return a, b, paths

import asyncio, httpx
from prefect import task

@task
async def download_many(urls: list[str]) -> list[str]:
    limits = httpx.Limits(max_connections=10)
    timeout = httpx.Timeout(60)
    sem = asyncio.Semaphore(5)

    async with httpx.AsyncClient(limits=limits, timeout=timeout) as client:
        async def fetch(u):
            async with sem:
                path = f"data/{u.split('/')[-1]}"
                async with client.stream("GET", u) as r:
                    r.raise_for_status()
                    with open(path, "wb") as f:
                        async for chunk in r.aiter_bytes():
                            f.write(chunk)
                return path

        return await asyncio.gather(*[fetch(u) for u in urls])

from pathlib import Path
import httpx

def ensure_dir(path: str | Path) -> Path:
    p = Path(path)
    p.mkdir(parents=True, exist_ok=True)
    return p

def download_stream(url: str, dest_path: str | Path, timeout: float = 60.0) -> str:
    dest_path = Path(dest_path)
    ensure_dir(dest_path.parent)
    with httpx.stream("GET", url, timeout=timeout) as r:
        r.raise_for_status()
        with open(dest_path, "wb") as f:
            for chunk in r.iter_bytes():
                f.write(chunk)
    return str(dest_path)

from pathlib import Path
from prefect import task, get_run_logger
from <http://utils.io|utils.io> import download_stream

# Replace these with your real endpoints
TSE_ENDPOINTS = {
    "tse_a": "<https://example.com/tse/a.csv>",
    "tse_b": "<https://example.com/tse/b.csv>",
    "tse_c": "<https://example.com/tse/c.csv>",
}

@task(retries=2, retry_delay_seconds=5, timeout_seconds=120, log_prints=True)
def download_tse_endpoint(name: str, url: str, out_dir: str = "data/tse") -> str:
    logger = get_run_logger()
    dest = Path(out_dir) / f"{name}.csv"
    <http://logger.info|logger.info>(f"Downloading TSE endpoint '{name}' -> {dest}")
    return download_stream(url, dest)

from pathlib import Path
from prefect import task, get_run_logger
from <http://utils.io|utils.io> import download_stream

# Replace with your real manifest URL
CONGRESSO_MANIFEST_URL = "<https://example.com/congresso/manifest.json>"

@task(retries=2, retry_delay_seconds=5, timeout_seconds=120, log_prints=True)
def download_manifest(out_dir: str = "data/congresso") -> str:
    logger = get_run_logger()
    dest = Path(out_dir) / "manifest.json"
    <http://logger.info|logger.info>(f"Downloading Congresso manifest -> {dest}")
    return download_stream(CONGRESSO_MANIFEST_URL, dest)

from pathlib import Path
from prefect import task, get_run_logger
from <http://utils.io|utils.io> import download_stream

# Replace these with your real endpoints
CONGRESSO_A_ENDPOINTS = {
    "a_1": "<https://example.com/congresso/a_1.json>",
    "a_2": "<https://example.com/congresso/a_2.json>",
    "a_3": "<https://example.com/congresso/a_3.json>",
}

@task(retries=2, retry_delay_seconds=5, timeout_seconds=120, log_prints=True)
def download_congresso_a_endpoint(name: str, url: str, manifest_path: str, out_dir: str = "data/congresso/a") -> str:
    # Use manifest_path if you need to read metadata for requests
    logger = get_run_logger()
    dest = Path(out_dir) / f"{name}.json"
    <http://logger.info|logger.info>(f"[A] Downloading {name} using manifest {manifest_path} -> {dest}")
    return download_stream(url, dest)

flows/pipeline.py

Why this matches your requirements
- One flow orchestrates everything.
- TSE endpoints run in parallel.
- Congresso downloads the manifest first, then runs Subtask A (parallel) and Subtask B (sequential) at the same time.
- All downloads stream to disk; tasks return file paths, not large payloads.
- ThreadPoolTaskRunner is ideal for IO-bound HTTP work.

Tuning and tips
- Control parallelism by adjusting `max_workers` in ThreadPoolTaskRunner.
- Add rate limiting if you need it later (Prefect concurrency limits or a simple `asyncio.Semaphore` inside a task if you switch to async).
- Put secrets (API keys) in Secret blocks and read them in tasks — avoid Variables for sensitive data.
- For production, consider persisting only small results and store big files in object storage; return URIs instead of local paths.

Run it
- Start your local server (optional if you just want local logs/observability):

- Execute the flow:

- Or serve/schedule locally:

download_stream

from __future__ import annotations

import asyncio
import json
from pathlib import Path
import hashlib
import httpx

def ensure_dir(path: str | Path) -> Path:
    p = Path(path)
    p.mkdir(parents=True, exist_ok=True)
    return p

def download_stream(url: str, dest_path: str | Path, timeout: float = 60.0) -> str:
    dest_path = Path(dest_path)
    ensure_dir(dest_path.parent)
    with httpx.stream("GET", url, timeout=timeout) as r:
        r.raise_for_status()
        with open(dest_path, "wb") as f:
            for chunk in r.iter_bytes():
                f.write(chunk)
    return str(dest_path)

def fetch_json(url: str, timeout: float = 30.0) -> dict:
    with httpx.Client(timeout=timeout) as client:
        r = client.get(url)
        r.raise_for_status()
        return r.json()

def save_json(url: str, dest_path: str | Path, timeout: float = 30.0) -> str:
    dest_path = Path(dest_path)
    ensure_dir(dest_path.parent)
    with httpx.Client(timeout=timeout) as client:
        r = client.get(url)
        r.raise_for_status()
        data = r.json()
    with open(dest_path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False)
    return str(dest_path)

def write_ndjson(records: list[dict], dest_path: str | Path) -> str:
    dest_path = Path(dest_path)
    ensure_dir(dest_path.parent)
    with open(dest_path, "w", encoding="utf-8") as f:
        for rec in records:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    return str(dest_path)

async def fetch_json_many_async(
    urls: list[str],
    out_dir: str | Path | None = None,
    concurrency: int = 10,
    timeout: float = 30.0,
) -> list[str] | list[dict]:
    """
    - If out_dir is provided, saves each JSON to a file and returns list of file paths.
    - Otherwise returns list of parsed dicts in memory.
    """
    sem = asyncio.Semaphore(concurrency)
    limits = httpx.Limits(max_connections=max(concurrency, 10))
    timeout_cfg = httpx.Timeout(timeout)
    ensure_dir(out_dir) if out_dir else None

    async def one(u: str):
        async with sem:
            async with httpx.AsyncClient(limits=limits, timeout=timeout_cfg) as client:
                r = await client.get(u)
                r.raise_for_status()
                data = r.json()
            if out_dir:
                # deterministic filename by URL hash
                name = hashlib.sha1(u.encode()).hexdigest() + ".json"
                path = Path(out_dir) / name
                with open(path, "w", encoding="utf-8") as f:
                    json.dump(data, f, ensure_ascii=False)
                return str(path)
            else:
                return data

    return await asyncio.gather(*[one(u) for u in urls])

tasks/congresso/base.py (single JSON manifest)

[tse]
base_url = "<https://api.tse.example>"
stream_timeout = 120

[congresso]
base_url = "<https://api.congresso.example>"
json_timeout = 30
json_concurrency = 10

[runner]
max_workers = 16

import os, json
from pathlib import Path
from dataclasses import dataclass
from typing import Any
from prefect.variables import Variable

try:
    import tomllib  # Python 3.11+
except ModuleNotFoundError:
    import tomli as tomllib  # pip install tomli for 3.10-

@dataclass
class Settings:
    tse_base_url: str
    tse_stream_timeout: float
    congresso_base_url: str
    congresso_json_timeout: float
    congresso_json_concurrency: int
    runner_max_workers: int

def _read_toml(path: str | Path) -> dict[str, Any]:
    p = Path(path)
    if not p.exists():
        return {}
    with open(p, "rb") as f:
        return tomllib.load(f)

def _get_var(name: str) -> str | None:
    # Prefect Variables override if set (non-sensitive only)
    try:
        v = Variable.get(name)
        return None if v in ("", None) else v
    except Exception:
        return None

def load_settings(config_path: str | Path = "config/settings.toml") -> Settings:
    cfg = _read_toml(config_path)

    def pick(var_name: str, env_name: str, default: Any) -> Any:
        # priority: Prefect Variable -> ENV -> TOML -> default
        v = _get_var(var_name)
        if v is not None:
            return v
        if (e := os.getenv(env_name)) is not None:
            return e
        return default

    return Settings(
        tse_base_url=str(pick("TSE_BASE_URL", "MYAPP_TSE_BASE_URL", cfg.get("tse", {}).get("base_url", ""))),
        tse_stream_timeout=float(pick("TSE_STREAM_TIMEOUT", "MYAPP_TSE_STREAM_TIMEOUT", cfg.get("tse", {}).get("stream_timeout", 120))),
        congresso_base_url=str(pick("CONGRESSO_BASE_URL", "MYAPP_CONGRESSO_BASE_URL", cfg.get("congresso", {}).get("base_url", ""))),
        congresso_json_timeout=float(pick("CONGRESSO_JSON_TIMEOUT", "MYAPP_CONGRESSO_JSON_TIMEOUT", cfg.get("congresso", {}).get("json_timeout", 30))),
        congresso_json_concurrency=int(pick("CONGRESSO_JSON_CONCURRENCY", "MYAPP_CONGRESSO_JSON_CONCURRENCY", cfg.get("congresso", {}).get("json_concurrency", 10))),
        runner_max_workers=int(pick("RUNNER_MAX_WORKERS", "MYAPP_RUNNER_MAX_WORKERS", cfg.get("runner", {}).get("max_workers", 16))),
    )

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner
from config.loader import load_settings

@flow
def pipeline():
    s = load_settings()
    # You can pass these into tasks or construct runners with them
    print("Using:", s)

if __name__ == "__main__":
    s = load_settings()
    pipeline.with_options(task_runner=ThreadPoolTaskRunner(max_workers=s.runner_max_workers))()

uv run prefect variable set TSE_BASE_URL <https://api.tse.example>
uv run prefect variable set RUNNER_MAX_WORKERS 32

uv run prefect variable unset RUNNER_MAX_WORKERS -y

Variable.get(...)

StrictVariableValue

str | None

str | None

Variable.get

await

from typing import Optional, cast
from prefect.variables import Variable
from prefect.types import StrictVariableValue

def get_var(name: str) -> Optional[StrictVariableValue]:
    # Cast removes the Sync[...] union for type-checkers
    v = cast(StrictVariableValue, Variable.get(name, default=None))
    return v

def get_var_str(name: str) -> str | None:
    v = get_var(name)
    return None if v is None else str(v)

def get_var_int(name: str, default: int) -> int:
    v = get_var(name)
    if v is None:
        return default
    if isinstance(v, bool):
        return int(v)
    if isinstance(v, (int, float, str)):
        return int(v)
    raise TypeError(f"Unsupported type for int: {type(v)}")

from typing import Optional
from prefect.variables import Variable
from prefect.types import StrictVariableValue

async def aget_var(name: str) -> Optional[StrictVariableValue]:
    return await Variable.get(name, default=None)

async def aget_var_str(name: str) -> str | None:
    v = await aget_var(name)
    return None if v is None else str(v)

cast(StrictVariableValue, Variable.get(...))

from typing import Optional, cast
from prefect.variables import Variable
from prefect.types import StrictVariableValue

def _get_var(name: str) -> Optional[StrictVariableValue]:
    v = cast(StrictVariableValue, Variable.get(name, default=None))
    return None if v in ("", None) else v

return Settings(
    runner_max_workers=int(pick("RUNNER_MAX_WORKERS","MYAPP_RUNNER_MAX_WORKERS", cfg.get("runner", {}).get("max_workers", 16))),

    tse_base_url=str(pick("TSE_BASE_URL", "MYAPP_TSE_BASE_URL", cfg.get("tse", {}).get("base_url", ""))),
    tse_stream_timeout=float(pick("TSE_STREAM_TIMEOUT", "MYAPP_TSE_STREAM_TIMEOUT", cfg.get("tse", {}).get("stream_timeout", 120))),

    congresso_base_url=str(pick("CONGRESSO_BASE_URL","MYAPP_CONGRESSO_BASE_URL", cfg.get("congresso", {}).get("base_url", ""))),
    congresso_json_timeout=float(pick("CONGRESSO_JSON_TIMEOUT","MYAPP_CONGRESSO_JSON_TIMEOUT", cfg.get("congresso", {}).get("json_timeout", 30))),
    congresso_json_concurrency=int(pick("CONGRESSO_JSON_CONCURRENCY","MYAPP_CONGRESSO_JSON_CONCURRENCY", cfg.get("congresso", {}).get("json_concurrency", 10))),
)

def pick_str(var, env, default): 
    v = _get_var(var); 
    if v is not None: return str(v)
    e = os.getenv(env)
    return e if e is not None else str(default)

def pick_int(var, env, default):
    v = _get_var(var)
    if v is not None:
        if isinstance(v, bool):  # avoid True->1 surprises
            raise TypeError(f"{var} should be int, got bool")
        return int(v)
    e = os.getenv(env)
    if e is not None: return int(e)
    return int(default)

def pick_float(var, env, default):
    v = _get_var(var)
    if v is not None:
        if isinstance(v, bool):
            raise TypeError(f"{var} should be float, got bool")
        return float(v)
    e = os.getenv(env)
    if e is not None: return float(e)
    return float(default)

[tse]
base_url = "<https://api.tse.example>"
stream_timeout = 120

[congresso]
base_url = "<https://api.congresso.example>"
json_timeout = 30
json_concurrency = 10

[runner]
max_workers = 16

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any

try:
    import tomllib  # Python 3.11+
except ModuleNotFoundError:
    import tomli as tomllib  # pip install tomli for 3.10

@dataclass
class Settings:
    # TSE
    tse_base_url: str
    tse_stream_timeout: float = 120.0

    # CONGRESSO
    congresso_base_url: str
    congresso_json_timeout: float = 30.0
    congresso_json_concurrency: int = 10

    # Runner
    runner_max_workers: int = 16

def _read_toml(path: str | Path) -> dict[str, Any]:
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"Config file not found: {p.resolve()}")
    with open(p, "rb") as f:
        return tomllib.load(f)

def load_settings(config_path: str | Path = "config/settings.toml") -> Settings:
    cfg = _read_toml(config_path)

    # Required keys (fail fast with clear errors)
    try:
        tse_cfg = cfg["tse"]
        congresso_cfg = cfg["congresso"]
    except KeyError as e:
        raise ValueError(f"Missing section in TOML: {e}. Expected [tse], [congresso].") from e

    try:
        tse_base_url = str(tse_cfg["base_url"])
        congresso_base_url = str(congresso_cfg["base_url"])
    except KeyError as e:
        raise ValueError(f"Missing required key: {e}. Ensure base_url is set.") from e

    # Optional keys with defaults from the dataclass
    tse_stream_timeout = float(tse_cfg.get("stream_timeout", Settings.tse_stream_timeout))
    congresso_json_timeout = float(congresso_cfg.get("json_timeout", Settings.congresso_json_timeout))
    congresso_json_concurrency = int(congresso_cfg.get("json_concurrency", Settings.congresso_json_concurrency))
    runner_max_workers = int(cfg.get("runner", {}).get("max_workers", Settings.runner_max_workers))

    return Settings(
        tse_base_url=tse_base_url,
        tse_stream_timeout=tse_stream_timeout,
        congresso_base_url=congresso_base_url,
        congresso_json_timeout=congresso_json_timeout,
        congresso_json_concurrency=congresso_json_concurrency,
        runner_max_workers=runner_max_workers,
    )

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner
from config.loader import load_settings

@flow
def pipeline():
    s = load_settings()
    # pass settings to tasks as needed
    return {"runner": s.runner_max_workers}

if __name__ == "__main__":
    s = load_settings()
    pipeline.with_options(task_runner=ThreadPoolTaskRunner(max_workers=s.runner_max_workers))()

<http://logger.info|logger.info>("Iniciando pipeline")

<http://logger.info|logger.info>

@flow

with_options(...)

pipeline

@flow

runner_max_workers

max_workers

s.max_workers

from prefect import flow, get_run_logger
from prefect.task_runners import ThreadPoolTaskRunner
from config.loader import load_settings

@flow(log_prints=True)  # no runner here; we’ll set it at call time
def pipeline():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Iniciando pipeline")

if __name__ == "__main__":
    s = load_settings()
    runner = ThreadPoolTaskRunner(max_workers=s.runner_max_workers)
    pipeline.with_options(task_runner=runner)()

with_options

from typing import cast
from prefect.flows import Flow

cast(Flow, pipeline).with_options(task_runner=runner)()
# or: pipeline.with_options(task_runner=runner)()  # type: ignore[attr-defined]

from prefect import flow, get_run_logger
from prefect.task_runners import ThreadPoolTaskRunner
from config.loader import load_settings

s = load_settings()
runner = ThreadPoolTaskRunner(max_workers=s.runner_max_workers)

@flow(task_runner=runner, log_prints=True)
def pipeline():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Iniciando pipeline")

if __name__ == "__main__":
    pipeline()

runner_max_workers

with_options

<http://logger.info|logger.info>("Iniciando pipeline")

from prefect import flow, get_run_logger
from prefect.task_runners import ThreadPoolTaskRunner
from config.loader import load_settings

s = load_settings()
runner = ThreadPoolTaskRunner(max_workers=s.runner_max_workers)  # ensure field name matches Settings

@flow(task_runner=runner, log_prints=True)  # mypy/pylance may complain, but this runs
def pipeline():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Iniciando pipeline")

if __name__ == "__main__":
    pipeline()

@flow(task_runner=runner, log_prints=True)  # type: ignore[call-arg]

from typing import cast
from prefect import flow, get_run_logger
from prefect.flows import Flow
from prefect.task_runners import ThreadPoolTaskRunner
from config.loader import load_settings

@flow(log_prints=True)
def pipeline():
    logger = get_run_logger()
    <http://logger.info|logger.info>("Iniciando pipeline")

if __name__ == "__main__":
    s = load_settings()
    runner = ThreadPoolTaskRunner(max_workers=s.runner_max_workers)
    cast(Flow, pipeline).with_options(task_runner=runner)()

uv run prefect version

from __future__ import annotations
from pathlib import Path
import json
import httpx

def ensure_dir(path: str | Path) -> Path:
    p = Path(path)
    p.mkdir(parents=True, exist_ok=True)
    return p

def filename_from_url(url: str) -> str:
    # naive: last path segment
    return url.rstrip("/").split("/")[-1] or "download.bin"

def download_stream(url: str, dest_path: str | Path, timeout: float = 120.0) -> str:
    dest_path = Path(dest_path)
    ensure_dir(dest_path.parent)
    with httpx.stream("GET", url, timeout=timeout) as r:
        r.raise_for_status()
        with open(dest_path, "wb") as f:
            for chunk in r.iter_bytes():
                f.write(chunk)
    return str(dest_path)

def fetch_json(url: str, timeout: float = 30.0) -> dict:
    with httpx.Client(timeout=timeout) as client:
        r = client.get(url)
        r.raise_for_status()
        return r.json()

def save_json(url: str, dest_path: str | Path, timeout: float = 60.0) -> str:
    data = fetch_json(url, timeout=timeout)
    dest_path = Path(dest_path)
    ensure_dir(dest_path.parent)
    with open(dest_path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False)
    return str(dest_path)

from pathlib import Path
from prefect import task, get_run_logger
from <http://utils.io|utils.io> import download_stream, filename_from_url

# Fill this with your 30 URLs
TSE_ENDPOINTS: list[str] = [
    # "https://.../file1.zip",
    # ...
]

@task(retries=2, retry_delay_seconds=5, timeout_seconds=600, log_prints=True)
def download_tse_zip(url: str, out_dir: str = "data/tse") -> str:
    logger = get_run_logger()
    name = filename_from_url(url)
    dest = Path(out_dir) / name
    <http://logger.info|logger.info>(f"TSE: downloading {url} -> {dest}")
    return download_stream(url, dest)

from prefect import task, get_run_logger
from <http://utils.io|utils.io> import fetch_json

# Replace with real endpoint
LEGISLATURA_URL = "<https://api.congresso.example/legislatura>"

@task(retries=2, retry_delay_seconds=5, timeout_seconds=60, log_prints=True)
def fetch_legislatura() -> dict:
    logger = get_run_logger()
    <http://logger.info|logger.info>("Congresso: fetching Legislatura")
    return fetch_json(LEGISLATURA_URL, timeout=60.0)

from pathlib import Path
from prefect import task, get_run_logger
from <http://utils.io|utils.io> import save_json

# Template; build the actual URL from legislatura fields
def deputados_url(legislatura: dict) -> str:
    # e.g., f"<https://api.congresso.example/deputados?legislatura={legislatura['id']}>"
    return "<https://api.congresso.example/deputados>"

@task(retries=2, retry_delay_seconds=5, timeout_seconds=120, log_prints=True)
def fetch_deputados(legislatura: dict, out_dir: str = "data/congresso") -> str:
    logger = get_run_logger()
    url = deputados_url(legislatura)
    dest = Path(out_dir) / "deputados.json"
    <http://logger.info|logger.info>(f"Congresso: fetching Deputados from {url} -> {dest}")
    return save_json(url, dest, timeout=60.0)

flows/pipeline.py
- Orchestrate everything; TSE downloads run in parallel; Legislatura first, then Deputados and Frentes in parallel

only

from datetime import timedelta
from prefect import flow, task
from prefect.cache_policies import CachePolicy

@task(
    retries=2,
    retry_delay_seconds=10,
    cache_policy=CachePolicy(),
    cache_expiration=timedelta(days=1),
)
def call_endpoint(name: str):
    # call your endpoint here
    ...

@flow
def sync_all(endpoints: list[str], only: list[str] | None = None):
    targets = [e for e in endpoints if not only or e in only]
    # Either map or loop — both are fine
    call_endpoint.map(targets)

sync_all(endpoints=ALL_ENDPOINTS)

sync_all(endpoints=ALL_ENDPOINTS, only=["/users", "/orders"])

CachePolicy

refresh_cache=True

from prefect import flow

@flow
def sync_one(endpoint: str):
    call_endpoint(endpoint)

@flow
def sync_all(endpoints: list[str], only: list[str] | None = None):
    for e in endpoints:
        if not only or e in only:
            sync_one(e)

only

prefect deployment run my-flow/my-deployment -p only='["/users","/orders"]' --watch

@task

name=

task_run_name=

from prefect import task, flow

@task(
    name="Download TSE endpoint",
    task_run_name="Download {endpoint}"   # uses the task parameter names
)
def download_tse_endpoint(endpoint: str):
    ...

@flow
def sync(endpoints: list[str]):
    # Each mapped run is named "Download <endpoint>"
    download_tse_endpoint.map(endpoints)

download_tse_endpoint.with_options(
    task_run_name="Download {endpoint}"
).map(endpoints)

def run_name(params: dict) -> str:
    ep = params.get("endpoint", "unknown")
    return f"Download {ep}"

@task(task_run_name=run_name)
def download_tse_endpoint(endpoint: str):
    ...

task_run_name

-252

@flow(flow_run_name="Sync {date}")

task_run_name

task_run_name

name

from pathlib import Path
from prefect import task, get_run_logger

@task(
    name="Fetch TSE endpoint",                 # static node label
    task_run_name="fetch_tse_{name}",          # dynamic run name per call
    retries=5, retry_delay_seconds=5, timeout_seconds=120, log_prints=True
)
def fetch_tse(name: str, url: str, out_dir: str = "data/tse") -> str:
    logger = get_run_logger()
    dest = Path(out_dir) / f"{name}.csv"
    <http://logger.info|logger.info>(f"Fazendo download do endpoint TSE '{name}' -> {dest}")
    return download_stream(url, dest)

name

url

from prefect import flow, unmapped

@flow
def sync_all():
    names, urls = zip(*TSE_ENDPOINTS.items())   # preserves pairing
    fetch_tse.map(name=names, url=urls, out_dir=unmapped("data/tse"))

from prefect import unmapped

names, urls = zip(*TSE_ENDPOINTS.items())
fetch_tse.with_options(task_run_name="fetch_tse_{name}").map(
    name=names, url=urls, out_dir=unmapped("data/tse")
)

task_run_name

name

url

-252

def rn(params: dict) -> str:
    return f"fetch_tse_{params['name']}"
fetch_tse = fetch_tse.with_options(task_run_name=rn)

task_run_name

pipeline

flows/pipeline.py

main.py

if __name__ == "__main__": pipeline()

flows/pipeline.py

# From the root of your repo (where ./flows/pipeline.py exists)
prefect deploy ./flows/pipeline.py:pipeline -n pipeline-dev

prefect worker start -p <your-work-pool>

# Run now and watch logs
prefect deployment run pipeline/pipeline-dev --watch

# Pass parameters if needed (example):
prefect deployment run pipeline/pipeline-dev -p date_today='"2025-10-21"'

ENTRYPOINT

path/to/file.py:flow_function_name

./flows/pipeline.py:pipeline

from prefect import flow

flow.from_source(
    "<https://github.com/owner/repo>",
    entrypoint="flows/pipeline.py:pipeline",
).deploy(name="pipeline-dev", work_pool_name="<pool-name>")

prefect deploy

pipeline

./flows/pipeline.py

@flow def pipeline(...)

prefect work-pool create "process-pool" --type process

path/to/file.py:flow_function_name

prefect deploy ./flows/pipeline.py:pipeline -n pipeline-dev

process-pool

prefect worker start -p process-pool

prefect deployment run pipeline/pipeline-dev --watch

date_today

prefect deployment run pipeline/pipeline-dev -p date_today='"2025-10-22"' --watch

./flows/pipeline.py:pipeline

main.py

ThreadPoolTaskRunner

--cron

--interval

from flows.pipeline import pipeline

if __name__ == "__main__":
    # Creates a deployment and starts a local runner that polls for runs
    pipeline.serve(
        name="deploy-1",
        # optional: schedule it
        # interval=3600,  # run every hour
        # or cron="0 7 * * *",
    )

uv run src/main.py

prefect deployment run pipeline/deploy-1 --watch

pause_on_shutdown=False

prefect work-pool create process-pool --type process
prefect deploy ./flows/pipeline.py:pipeline -n deploy-1
prefect worker start -p process-pool
prefect deployment run pipeline/deploy-1 --watch

pipeline()

pipeline.serve(...)

if __name__ == "__main__": pipeline.serve(...)

prefect deployment run ...

INPUTS

CachePolicy()

@task(cache_policy=...)

[tse]
retries = 5
retry_delay = 5
timeout = 120
cache_policy = "INPUTS"          # or "NONE", "DEFAULT", "FLOW_PARAMETERS", "TASK_SOURCE", "RUN_ID"
cache_expiration_seconds = 86400 # optional, if you also set cache_expiration

from datetime import timedelta
from prefect.cache_policies import (
    CachePolicy,
    INPUTS,
    NONE,          # aka NO_CACHE in some contexts
    DEFAULT,
    TASK_SOURCE,
    FLOW_PARAMETERS,
    RUN_ID,
)

_CACHE_POLICY_MAP = {
    "INPUTS": INPUTS,
    "NONE": NONE,
    "NO_CACHE": NONE,
    "DEFAULT": DEFAULT,
    "TASK_SOURCE": TASK_SOURCE,
    "FLOW_PARAMETERS": FLOW_PARAMETERS,
    "RUN_ID": RUN_ID,
}

def coerce_cache_policy(policy_name: str | None):
    if policy_name is None:
        return None  # means "NotSet" if you don’t pass cache_policy
    if isinstance(policy_name, str):
        key = policy_name.strip().upper()
        if key in _CACHE_POLICY_MAP:
            return _CACHE_POLICY_MAP[key]
        raise ValueError(f"Unknown cache_policy '{policy_name}'. "
                         f"Expected one of: {', '.join(_CACHE_POLICY_MAP)}")
    # You could also allow pre-built CachePolicy objects here if needed
    raise TypeError(f"cache_policy must be a string or None, got {type(policy_name)}")

class TSEConfig(BaseModel):
    RETRIES: int
    RETRY_DELAY: int
    TIMEOUT: int
    CACHE_POLICY: str | None = None
    CACHE_EXPIRATION_SECONDS: int | None = None

    @property
    def cache_policy_obj(self):
        return coerce_cache_policy(self.CACHE_POLICY)

    @property
    def cache_expiration_td(self):
        return timedelta(seconds=self.CACHE_EXPIRATION_SECONDS) if self.CACHE_EXPIRATION_SECONDS else None

APP_SETTINGS = loader.load_config()

@task(
    retries=APP_SETTINGS.TSE.RETRIES,
    retry_delay_seconds=APP_SETTINGS.TSE.RETRY_DELAY,
    timeout_seconds=APP_SETTINGS.TSE.TIMEOUT,
    log_prints=True,
    cache_policy=APP_SETTINGS.TSE.cache_policy_obj,
    cache_expiration=APP_SETTINGS.TSE.cache_expiration_td,
    task_run_name="fetch_tse_{name}",
)
def extract_tse(name: str, url: str, out_dir: str = "data/tse") -> str:
    logger = get_run_logger()
    dest = Path(out_dir) / f"{name}.csv"
    <http://logger.info|logger.info>(f"Fazendo download do endpoint TSE '{name}' -> {dest}")
    return download_stream(url, dest)

task(cache_policy=...)

CachePolicy

INPUTS

CachePolicy

key_storage

isolation_level

CachePolicy(...)

cache_policy = "CUSTOM"

"data/tse/candidatos_2018.csv"

"<s3://bucket/key.csv>"

cache_policy=INPUTS

cache_expiration

persist_result=True

PREFECT_RESULTS_PERSIST_BY_DEFAULT=true

from datetime import timedelta
from pathlib import Path
from prefect import task, get_run_logger
from prefect.cache_policies import INPUTS

@task(
    retries=APP_SETTINGS.TSE.RETRIES,
    retry_delay_seconds=APP_SETTINGS.TSE.RETRY_DELAY,
    timeout_seconds=APP_SETTINGS.TSE.TIMEOUT,
    log_prints=True,
    persist_result=True,                  # make cache reusable across runs
    cache_policy=INPUTS,                  # cache by inputs (name, url, out_dir)
    cache_expiration=timedelta(days=30),  # optional TTL for freshness
    task_run_name="fetch_tse_{name}",
)
def extract_tse(name: str, url: str, out_dir: str = "data/tse") -> str:
    logger = get_run_logger()
    dest = Path(out_dir) / f"{name}.csv"

    # Local idempotency guard: if file already exists, skip network download
    if dest.exists() and dest.stat().st_size > 0:
        <http://logger.info|logger.info>(f"Arquivo já existe, reutilizando: {dest}")
        return str(dest)

    <http://logger.info|logger.info>(f"Fazendo download do endpoint TSE '{name}' -> {dest}")
    return download_stream(url, dest)

cache_key_fn

name

task_run_name="extract_tse_{name}"

from pathlib import Path
from prefect import task, get_run_logger
from datetime import timedelta

from <http://utils.io|utils.io> import download_stream
from config.loader import load_config, CACHE_POLICY_MAP

APP_SETTINGS = load_config()

@task(
    retries=APP_SETTINGS.TSE.RETRIES,
    retry_delay_seconds=APP_SETTINGS.TSE.RETRY_DELAY,
    timeout_seconds=APP_SETTINGS.TSE.TIMEOUT,
    log_prints=True,
    cache_policy=CACHE_POLICY_MAP[APP_SETTINGS.TSE.CACHE_POLICY],
    cache_expiration=timedelta(days=APP_SETTINGS.TSE.CACHE_EXPIRATION),
    task_run_name="extract_tse_{name}",   # <- dynamic run name per invocation
)
def extract_tse(name: str, url: str, out_dir: str = "data/tse") -> str:
    logger = get_run_logger()
    dest = Path(out_dir) / f"{name}.csv"
    <http://logger.info|logger.info>(f"Fazendo download do endpoint TSE '{name}' -> {dest}")
    return download_stream(url, dest)

# inside your flow
for key, url in TSE_ENDPOINTS.items():
    extract_tse.submit(name=key, url=url)

extract_tse_<key>-<suffix>

extract_tse_candidatos_2018-34b

task_run_name

name

url

{name}

-34b

extract_tse.with_options(task_run_name="extract_tse_{name}").submit(name=key, url=url)

def rn(params: dict) -> str:
    raw = params["name"]
    short = raw[:60]
    return f"extract_tse_{short}"

extract_tse = extract_tse.with_options(task_run_name=rn)

from datetime import date, timedelta
import re

CURRENT_YEAR = date.today().year

def extract_year_from_name(name: str) -> int | None:
    # e.g., "candidatos_2018", "redes_sociais_2024_SP"
    m = re.search(r"\b(20\d{2})\b", name)
    return int(m.group(1)) if m else None

def cache_opts_for_year(year: int | None):
    if year is None:
        # no year in name -> default behavior
        return {}
    if year < CURRENT_YEAR:
        # cache past years
        return {
            "cache_policy": CACHE_POLICY_MAP["INPUTS"],
            "cache_expiration": timedelta(days=365*5),
            "persist_result": True,
        }
    else:
        # do not cache current year
        return {
            "cache_policy": CACHE_POLICY_MAP["NO_CACHE"],
            # OR: "refresh_cache": True,
        }

# inside your flow
for name, url in TSE_ENDPOINTS.items():
    year = extract_year_from_name(name)
    opts = cache_opts_for_year(year)
    extract_tse.with_options(**opts).submit(name=name, url=url)

@task(
    # defaults here are fine — we’ll override per call
    cache_policy=CACHE_POLICY_MAP[APP_SETTINGS.TSE.CACHE_POLICY],
    cache_expiration=timedelta(days=APP_SETTINGS.TSE.CACHE_EXPIRATION),
    task_run_name="extract_tse_{name}",
)
def extract_tse(name: str, url: str, year: int | None = None, out_dir: str = "data/tse") -> str:
    ...

for name, url in TSE_ENDPOINTS.items():
    year = extract_year_from_name(name)
    opts = cache_opts_for_year(year)
    extract_tse.with_options(**opts).submit(name=name, url=url, year=year)

[tse]
cache_policy = "INPUTS"              # default for past years
cache_expiration = 3650              # 10 years
no_cache_policy = "NO_CACHE"         # used for current year

def cache_opts_for_year(year):
    if year is None or year < CURRENT_YEAR:
        return {
            "cache_policy": CACHE_POLICY_MAP[APP_SETTINGS.TSE.CACHE_POLICY],
            "cache_expiration": timedelta(days=APP_SETTINGS.TSE.CACHE_EXPIRATION),
            "persist_result": True,
        }
    else:
        return {
            "cache_policy": CACHE_POLICY_MAP[APP_SETTINGS.TSE.NO_CACHE_POLICY],
        }

extract_year_from_name

APP_SETTINGS = load_config()

APP_SETTINGS = load_config()

cache_policy=CACHE_POLICY_MAP[APP_SETTINGS.TSE.CACHE_POLICY]

with_options(...)

from datetime import timedelta
from prefect import task, flow
from prefect.cache_policies import INPUTS, NO_CACHE

@task(
    retries=5,
    retry_delay_seconds=5,
    timeout_seconds=120,
    log_prints=True,
    task_run_name="extract_tse_{name}",
)
def extract_tse(name: str, url: str, out_dir: str="data/tse") -> str:
    ...

@flow
def pipeline(...):
    settings = load_config()  # read TOML per run
    cache_policy = CACHE_POLICY_MAP[settings.TSE.CACHE_POLICY]  # e.g. "NO_CACHE" -> NO_CACHE
    cache_exp = timedelta(days=settings.TSE.CACHE_EXPIRATION)

    for name, url in TSE_ENDPOINTS.items():
        extract_tse.with_options(
            cache_policy=cache_policy,
            cache_expiration=cache_exp,
            persist_result=True,   # if you want cross-run reuse
        ).submit(name=name, url=url)

@flow
def pipeline(cache_policy: str = "NO_CACHE", cache_expiration_days: int = 90):
    cache_policy_obj = CACHE_POLICY_MAP[cache_policy]
    cache_exp = timedelta(days=cache_expiration_days)
    ...

prefect deployment run pipeline/deploy-1 -p cache_policy='"INPUTS"' -p cache_expiration_days=365 --watch

from prefect.variables import Variable
policy_name = Variable.get("TSE_CACHE_POLICY", default="NO_CACHE")
exp_days = int(Variable.get("TSE_CACHE_EXPIRATION_DAYS", default="90"))

prefect deployment run pipeline/deploy-1 \
  -jv TSE_CACHE_POLICY='"INPUTS"' \
  -jv TSE_CACHE_EXPIRATION_DAYS=365 \
  --watch

@task(cache_policy=...)

with_options(...)

runner.process_limit

ThreadPoolTaskRunner(max_workers=16)

runner.process_limit

PREFECT_RUNNER_PROCESS_LIMIT

flow.serve(...)

runner.process_limit

flow.serve

runner.process_limit

max_workers

runner.process_limit

max_workers

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner

APP_SETTINGS = load_config()

@flow(
    task_runner=ThreadPoolTaskRunner(
        max_workers=APP_SETTINGS.FLOW.MAX_RUNNERS  # from your TOML
    ),
    log_prints=True,
)
def pipeline(...):
    ...

refresh_cache=True

with_options

extract_tse.with_options(refresh_cache=True).submit(name=..., url=...)

PREFECT_TASKS_REFRESH_CACHE=true

cache_policy=NO_CACHE

with_options(...)

from prefect.cache_policies import NO_CACHE
  extract_tse.with_options(cache_policy=NO_CACHE).submit(...)

PREFECT_TASKS_DEFAULT_NO_CACHE=true

PREFECT_TASKS_DISABLE_CACHING=true

extract_tse.with_options(cache_expiration=timedelta(days=1)).submit(...)

~/.prefect/storage/

PREFECT_LOCAL_STORAGE_PATH

CachePolicy(key_storage=...)

prefect cache clear

get_run_logger()

print()

print()

<http://logger.info|logger.info>(...)

get_run_logger()

log_prints

get_run_logger()

print(...)

log_prints

[logging]
level = "INFO"
log_prints = true

log_prints = "True"

get_run_logger()

log_prints

print()

log_prints = true

@task(log_prints=True)
def my_task():
    print("this will be captured even if global log_prints is false")

logging.log_prints

fetch_json_many_async

from pathlib import Path
from typing import cast
from prefect import task, get_run_logger

from <http://utils.io|utils.io> import fetch_json_many_async, save_ndjson
from config.loader import load_config

APP_SETTINGS = load_config()

def frentes_membros_urls(frentes_ids: list[str]) -> list[str]:
    return [f"{APP_SETTINGS.CAMARA.REST_BASE_URL}frentes/{id}/membros" for id in frentes_ids]

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=APP_SETTINGS.CAMARA.TIMEOUT,
)
async def extract_frentes_membros(frentes_ids: list[str], out_dir: str | Path = "data/camara") -> str:
    logger = get_run_logger()
    urls = frentes_membros_urls(frentes_ids)
    <http://logger.info|logger.info>(f"Congresso: buscando Membros de {len(urls)} Frentes")

    jsons = await fetch_json_many_async(
        urls,
        concurrency=APP_SETTINGS.CAMARA.CONCURRENCY,
        timeout=APP_SETTINGS.CAMARA.TIMEOUT,
        follow_pagination=True,
    )
    dest = Path(out_dir)
    return save_ndjson(cast(list[dict], jsons), dest)

fut = extract_frentes_membros.submit(...); result = fut.result()

result = await extract_frentes_membros(...)

from pathlib import Path
from typing import cast
from prefect import task, get_run_logger
from prefect.utilities.asyncutils import run_coro_as_sync

from <http://utils.io|utils.io> import fetch_json_many_async, save_ndjson
from config.loader import load_config

APP_SETTINGS = load_config()

def frentes_membros_urls(frentes_ids: list[str]) -> list[str]:
    return [f"{APP_SETTINGS.CAMARA.REST_BASE_URL}frentes/{id}/membros" for id in frentes_ids]

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=APP_SETTINGS.CAMARA.TIMEOUT,
)
def extract_frentes_membros(frentes_ids: list[str], out_dir: str | Path = "data/camara") -> str:
    logger = get_run_logger()
    urls = frentes_membros_urls(frentes_ids)
    <http://logger.info|logger.info>(f"Congresso: buscando Membros de {len(urls)} Frentes")

    jsons = run_coro_as_sync(
        fetch_json_many_async(
            urls,
            concurrency=APP_SETTINGS.CAMARA.CONCURRENCY,
            timeout=APP_SETTINGS.CAMARA.TIMEOUT,
            follow_pagination=True,
        )
    )
    dest = Path(out_dir)
    return save_ndjson(cast(list[dict], jsons), dest)

.submit()

.result()

run_coro_as_sync

from pathlib import Path
from typing import cast
from prefect import task, get_run_logger
from prefect.utilities.asyncutils import run_coro_as_sync

from <http://utils.io|utils.io> import fetch_json_many_async, save_ndjson
from config.loader import load_config

APP_SETTINGS = load_config()

def frentes_membros_urls(frentes_ids: list[str]) -> list[str]:
    return [f"{APP_SETTINGS.CAMARA.REST_BASE_URL}frentes/{id}/membros" for id in frentes_ids]

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=APP_SETTINGS.CAMARA.TIMEOUT,
)
def extract_frentes_membros(frentes_ids: list[str], out_dir: str | Path = "data/camara") -> str:
    logger = get_run_logger()
    urls = frentes_membros_urls(frentes_ids)
    <http://logger.info|logger.info>(f"Congresso: buscando Membros de {len(urls)} Frentes")

    # Bridge the async coroutine into this sync task
    jsons = run_coro_as_sync(
        fetch_json_many_async(
            urls,
            concurrency=APP_SETTINGS.CAMARA.CONCURRENCY,
            timeout=APP_SETTINGS.CAMARA.TIMEOUT,
            follow_pagination=True,
        )
    )
    dest = Path(out_dir)
    return save_ndjson(cast(list[dict], jsons), dest)

# ... your imports and flow definition

@flow(
    task_runner=ThreadPoolTaskRunner(max_workers=APP_SETTINGS.FLOW.MAX_RUNNERS),  # type: ignore
    log_prints=True,
)
def pipeline(
    date: date = datetime.now().date(),
    refresh_cache: bool = False,
):
    logger = get_run_logger()
    <http://logger.info|logger.info>("Iniciando pipeline")

    # TSE in parallel
    tse_fs = [
        # NOTE: pass function parameters to .submit; use with_options() only for task OPTIONS
        extract_tse.submit(name, url, refresh_cache=refresh_cache)
        for name, url in TSE_ENDPOINTS.items()
    ]

    # CONGRESSO
    legislatura = extract_legislatura(date)
    deputados_f = extract_deputados.submit(legislatura)
    frentes_f = extract_frentes.submit(legislatura)

    # If you have frentes_ids available, submit the async-backed task as usual
    # frentes_ids = ...  # derive this from your data
    # frentes_membros_f = extract_frentes_membros.submit(frentes_ids)

    # Gather results
    tse_paths = [f.result() for f in tse_fs]
    deputados_path = deputados_f.result()
    frentes_path = frentes_f.result()
    # frentes_membros_path = frentes_membros_f.result()

    <http://logger.info|logger.info>(f"Arquivos TSE: {len(tse_paths)}")
    <http://logger.info|logger.info>(f"Deputados -> {deputados_path}")
    <http://logger.info|logger.info>(f"Frentes -> {frentes_path}")
    # <http://logger.info|logger.info>(f"Frentes Membros -> {frentes_membros_path}")

    return {
        "tse": tse_paths,
        "congresso_deputados": deputados_path,
        "congresso_frentes": frentes_path,
        # "congresso_frentes_membros": frentes_membros_path,
    }

run_coro_as_sync

.submit(..., refresh_cache=refresh_cache)

with_options()

await

.submit()

await task(...)

from prefect import flow, get_run_logger
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import resolve_futures_to_results

@flow(
    task_runner=ThreadPoolTaskRunner(max_workers=APP_SETTINGS.FLOW.MAX_RUNNERS),  # type: ignore
    log_prints=True,
)
def pipeline(date: date = datetime.now().date(), refresh_cache: bool = False):
    logger = get_run_logger()
    <http://logger.info|logger.info>("Iniciando pipeline")

    # TSE in parallel
    tse_fs = [
        extract_tse.submit(name, url, refresh_cache=refresh_cache)
        for name, url in TSE_ENDPOINTS.items()
    ]

    # CONGRESSO
    legislatura = extract_legislatura(date)
    deputados_f = extract_deputados.submit(legislatura)
    frentes_f = extract_frentes.submit(legislatura)

    # If your task accepts the resolved frentes output, you can pass the future;
    # Prefect will resolve it as an upstream dependency
    frentes_membros_f = extract_frentes_membros.submit(frentes_f)

    # Resolve everything at once, preserving structure
    results = resolve_futures_to_results({
        "tse": tse_fs,
        "congresso_deputados": deputados_f,
        "congresso_frentes": frentes_f,
        "congresso_frentes_membros": frentes_membros_f,
    })

    <http://logger.info|logger.info>(f"Arquivos TSE: {len(results['tse'])}")
    <http://logger.info|logger.info>(f"Deputados -> {results['congresso_deputados']}")
    <http://logger.info|logger.info>(f"Frentes -> {results['congresso_frentes']}")
    <http://logger.info|logger.info>(f"Frentes Membros -> {results['congresso_frentes_membros']}")

    return results

.submit(...)

resolve_futures_to_results

from pathlib import Path
from typing import cast
from prefect import task, get_run_logger
from prefect.futures import PrefectFuture

@task(...)
def extract_frentes_membros(
    frentes_ids: list[str] | PrefectFuture[list[str]],
    out_dir: str | Path = "data/camara",
) -> str:
    ...

frentes_membros_f = extract_frentes_membros.submit(frentes_f)

from typing import Any, cast

frentes_membros_f = extract_frentes_membros.submit(cast(Any, frentes_f))

frentes_membros_f = extract_frentes_membros.submit(frentes_f)  # pyright: ignore[reportArgumentType]

.result()

from prefect.futures import resolve_futures_to_results

results = resolve_futures_to_results({
    "tse": tse_fs,
    "congresso_deputados": deputados_f,
    "congresso_frentes": frentes_f,
    "congresso_frentes_membros": frentes_membros_f,
})

PrefectFuture

extract_frentes_membros.submit(frentes_f)

list[str]

PrefectFuture[list[str]]

from prefect.futures import PrefectFuture

@task(...)
def extract_frentes_membros(
    frentes_ids: list[str] | PrefectFuture[list[str]],
    out_dir: str | Path = "data/camara",
) -> str:
    ...

from typing import Any, cast

frentes_membros_f = extract_frentes_membros.submit(cast(Any, frentes_f))

frentes_membros_f = extract_frentes_membros.submit(frentes_f)  # pyright: ignore[reportArgumentType]

fetch_json_many_async

from pathlib import Path
from typing import cast
from prefect import task, get_run_logger
from prefect.utilities.asyncutils import run_coro_as_sync

from <http://utils.io|utils.io> import fetch_json_many_async, save_ndjson
from config.loader import load_config

APP_SETTINGS = load_config()

def frentes_membros_urls(frentes_ids: list[str]) -> list[str]:
    return [f"{APP_SETTINGS.CAMARA.REST_BASE_URL}frentes/{id}/membros" for id in frentes_ids]

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=APP_SETTINGS.CAMARA.TIMEOUT,
)
def extract_frentes_membros(frentes_ids: list[str], out_dir: str | Path = "data/camara") -> str:
    logger = get_run_logger()
    urls = frentes_membros_urls(frentes_ids)
    <http://logger.info|logger.info>(f"Congresso: buscando Membros de {len(urls)} Frentes")

    jsons = run_coro_as_sync(
        fetch_json_many_async(
            urls,
            concurrency=APP_SETTINGS.CAMARA.CONCURRENCY,
            timeout=APP_SETTINGS.CAMARA.TIMEOUT,
            follow_pagination=True,
        )
    )
    dest = Path(out_dir)
    return save_ndjson(cast(list[dict], jsons), dest)

with_options

with_options

.result()

from prefect import flow, get_run_logger
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import resolve_futures_to_results

@flow(
    task_runner=ThreadPoolTaskRunner(max_workers=APP_SETTINGS.FLOW.MAX_RUNNERS),  # type: ignore
    log_prints=True,
)
def pipeline(date: date = datetime.now().date(), refresh_cache: bool = False):
    logger = get_run_logger()
    <http://logger.info|logger.info>("Iniciando pipeline")

    # TSE in parallel
    tse_fs = [
        extract_tse.submit(name, url, refresh_cache=refresh_cache)
        for name, url in TSE_ENDPOINTS.items()
    ]

    # CONGRESSO
    legislatura = extract_legislatura(date)
    deputados_f = extract_deputados.submit(legislatura)
    frentes_f = extract_frentes.submit(legislatura)

    # Passing a future to another task is fine; Prefect resolves it
    frentes_membros_f = extract_frentes_membros.submit(frentes_f)

    # Resolve everything at once (preserves structure)
    results = resolve_futures_to_results({
        "tse": tse_fs,
        "congresso_deputados": deputados_f,
        "congresso_frentes": frentes_f,
        "congresso_frentes_membros": frentes_membros_f,
    })

    return results

run_coro_as_sync

from pathlib import Path
from typing import cast
from prefect import task, get_run_logger

from <http://utils.io|utils.io> import fetch_json_many_async, save_ndjson
from config.loader import load_config

APP_SETTINGS = load_config()

def frentes_membros_urls(frentes_ids: list[str]) -> list[str]:
    return [f"{APP_SETTINGS.CAMARA.REST_BASE_URL}frentes/{id}/membros" for id in frentes_ids]

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=APP_SETTINGS.CAMARA.TIMEOUT,
)
async def extract_frentes_membros(frentes_ids: list[str], out_dir: str | Path = "data/camara") -> str:
    logger = get_run_logger()
    urls = frentes_membros_urls(frentes_ids)
    <http://logger.info|logger.info>(f"Congresso: buscando Membros de {len(urls)} Frentes")

    jsons = await fetch_json_many_async(
        urls,
        concurrency=APP_SETTINGS.CAMARA.CONCURRENCY,
        timeout=APP_SETTINGS.CAMARA.TIMEOUT,
        follow_pagination=True,
    )
    dest = Path(out_dir)
    return save_ndjson(cast(list[dict], jsons), dest)

frentes_membros_f = extract_frentes_membros.submit(frentes_f)

.result()

resolve_futures_to_results(...)

frentes_ids = await extract_frentes(legislatura)

result = await extract_frentes_membros(frentes_ids)

.submit()

.result()

def frentes_membros_urls(frentes_ids: list[str]) -> list[str]:
    return [f"{APP_SETTINGS.CAMARA.REST_BASE_URL}frentes/{id}/membros" for id in frentes_ids]

from typing import Any, cast
  frentes_membros_f = extract_frentes_membros.submit(cast(Any, frentes_f))

frentes_membros_f = extract_frentes_membros.submit(frentes_f)  # pyright: ignore[reportArgumentType]

from typing import Any, cast

  tse_fs = [
      cast(Any, extract_tse).submit(name, url, refresh_cache=refresh_cache)
      for name, url in TSE_ENDPOINTS.items()
  ]

from prefect.futures import resolve_futures_to_results

results = resolve_futures_to_results({
    "tse": tse_fs,
    "congresso_deputados": deputados_f,
    "congresso_frentes": frentes_f,
    "congresso_frentes_membros": frentes_membros_f,
})

from typing import Any
from prefect import get_run_logger
from prefect.exceptions import MissingContextError

def _get_prefect_logger_or_none() -> Any | None:
    try:
        return get_run_logger()
    except MissingContextError:
        return None

async def fetch_json_many_async(
    urls: list[str],
    out_dir: str | Path | None = None,
    concurrency: int = 10,
    timeout: float = 30.0,
    follow_pagination: bool = True,
    logger: Any | None = None,   # <- NEW
) -> list[str] | list[dict]:
    """
    - Se out_dir for fornecido, salva cada JSON em um arquivo e retorna a lista de caminhos
    - Caso contrário, retorna a lista de dicionários em memória
    """
    logger = logger or _get_prefect_logger_or_none()

    def log(msg: str):
        if logger:
            <http://logger.info|logger.info>(msg)
        else:
            print(msg)

    sem = asyncio.Semaphore(concurrency)
    limits = httpx.Limits(max_connections=max(concurrency, 10))
    timeout_cfg = httpx.Timeout(timeout)
    ensure_dir(out_dir) if out_dir else None

    processed_urls = set()
    results: list[str] | list[dict] = []

    async def one(u: str):
        if u in processed_urls:
            log(f"Skipping duplicate URL: {u}")
            return []
        processed_urls.add(u)

        log(f"Downloading {u}")
        async with sem:
            async with httpx.AsyncClient(limits=limits, timeout=timeout_cfg) as client:
                r = await client.get(u)
                r.raise_for_status()
                data = r.json()

        if out_dir:
            name = hashlib.sha1(u.encode()).hexdigest() + ".json"
            path = Path(out_dir) / name
            with open(path, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False)
            current_result = str(path)
            log(f"Downloaded {u} -> {current_result}")
        else:
            current_result = data
            log(f"Downloaded {u}")

        additional_results = []
        if follow_pagination and isinstance(data, dict) and "links" in data:
            links = {link["rel"]: link["href"] for link in data.get("links", []) if isinstance(link, dict)}
            if links.get("self") and links.get("last") and links["self"] != links["last"] and "next" in links:
                next_url = links["next"]
                log(f"Following pagination: {u} -> next {next_url}")
                additional_results = await one(next_url)

        if isinstance(additional_results, list):
            return [current_result] + additional_results
        else:
            return [current_result]

    tasks = [one(u) for u in urls]
    nested_results = await asyncio.gather(*tasks)

    for item in nested_results:
        if isinstance(item, list):
            results.extend(item)
        else:
            results.append(item)

    log(f"Finished: {len(results)} items")
    return results

@task(log_prints=False, ...)  # log_prints not needed when using logger
def my_task(...):
    logger = get_run_logger()
    jsons = run_coro_as_sync(fetch_json_many_async(urls, out_dir=None, logger=logger))

[logging]
    level = "INFO"
    log_prints = true

@task(log_prints=True)

print(f"Downloading {u}")
print(f"Downloaded {u} -> {current_result}")
print(f"Following pagination: {u} -> next {next_url}")
print(f"Finished: {len(results)} items")

prefect.artifacts.create_link_artifact

from prefect import flow
from prefect.artifacts import create_link_artifact

@flow
def demo():
    create_link_artifact(
        link="<https://dash.mycompany.com/reports/123>",
        link_text="Open KPI dashboard",    # optional; defaults to the URL if omitted
        key="kpi-dashboard-link",          # optional; lowercase letters, numbers, dashes
        description="Daily KPI dashboard for 2025-10-25",
    )

demo()

progress_artifact_id

if progress_artifact_id

key

from prefect import flow
from prefect.artifacts import acreate_progress_artifact, aupdate_progress_artifact

@flow
async def main(urls):
    # 1) Create the progress artifact and capture its UUID
    progress_id = await acreate_progress_artifact(
        progress=0,
        description="Starting downloads…",
        # Optional: add a key to also see it on the global Artifacts page
        key="camara-download-progress"
    )

    # 2) Pass the UUID to your async helper
    results = await fetch_json_many_async(
        urls=urls,
        concurrency=10,
        timeout=30.0,
        follow_pagination=True,
        progress_artifact_id=progress_id,
    )

    # 3) Final update to 100%
    await aupdate_progress_artifact(
        artifact_id=progress_id,
        progress=100,
        description="Downloads complete"
    )

if progress_artifact_id and len(urls) > 0:

aupdate_progress_artifact

key

progress_artifact_id

extract_frentes_membros

from prefect import task, get_run_logger
from prefect.artifacts import acreate_progress_artifact, aupdate_progress_artifact

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=APP_SETTINGS.CAMARA.TIMEOUT,
)
async def extract_frentes_membros(frentes_ids: list[str], out_dir: str | Path = "data/camara") -> str:
    logger = get_run_logger()

    # Create the progress artifact (optionally add a key to also see it on the global Artifacts page)
    progress_artifact = await acreate_progress_artifact(
        progress=0.0,
        description="Download de membros de frentes da Câmara",
        # key="frentes-membros-progress"  # optional
    )
    <http://logger.info|logger.info>(f"Progress artifact id: {progress_artifact}")

    urls = frentes_membros_urls(frentes_ids)
    <http://logger.info|logger.info>(f"Congresso: buscando Membros de {len(urls)} Frentes")

    jsons = await fetch_json_many_async(
        urls=urls,
        concurrency=APP_SETTINGS.CAMARA.CONCURRENCY,
        timeout=APP_SETTINGS.CAMARA.TIMEOUT,
        follow_pagination=True,
        progress_artifact_id=progress_artifact,
    )

    # Final update to 100% is a nice touch
    await aupdate_progress_artifact(
        artifact_id=progress_artifact,
        progress=100.0,
        description="Downloads concluídos"
    )

    dest = Path(out_dir) / "frentes_membros.ndjson"
    return save_ndjson(cast(list[dict], jsons), dest)

from prefect.artifacts import aupdate_progress_artifact

# ...
if progress_artifact_id and len(urls) > 0:
    async with update_lock:
        downloaded_urls += 1
        await aupdate_progress_artifact(
            artifact_id=progress_artifact_id,
            progress=(downloaded_urls / len(urls)) * 100
        )

update_progress_artifact

urls

<coroutine object acreate_progress_artifact at ...>

acreate_progress_artifact(...)

from uuid import UUID
from prefect.artifacts import acreate_progress_artifact, aupdate_progress_artifact

@task(...)
async def extract_frentes_membros(frentes_ids: list[str], out_dir: str | Path = "data/camara") -> str:
    logger = get_run_logger()

    progress_id = await acreate_progress_artifact(
        progress=0.0,
        description="Download de membros de frentes da Câmara",
        key="frentes-membros-progress"
    )
    <http://logger.info|logger.info>(f"Progress artifact id: {progress_id} (type={type(progress_id)})")
    assert isinstance(progress_id, UUID)

    urls = frentes_membros_urls(frentes_ids)
    <http://logger.info|logger.info>(f"Congresso: buscando Membros de {len(urls)} Frentes")

    jsons = await fetch_json_many_async(
        urls=urls,
        concurrency=APP_SETTINGS.CAMARA.CONCURRENCY,
        timeout=APP_SETTINGS.CAMARA.TIMEOUT,
        follow_pagination=True,
        progress_artifact_id=progress_id,  # <- pass the UUID
    )

    await aupdate_progress_artifact(
        artifact_id=progress_id,
        progress=100.0,
        description="Downloads concluídos"
    )

    dest = Path(out_dir) / "frentes_membros.ndjson"
    return save_ndjson(cast(list[dict], jsons), dest)

from prefect.artifacts import aupdate_progress_artifact

# ...
if progress_artifact_id and len(urls) > 0:
    async with update_lock:
        downloaded_urls += 1
        await aupdate_progress_artifact(
            artifact_id=progress_artifact_id,
            progress=(downloaded_urls / len(urls)) * 100
        )

aupdate_progress_artifact

progress_id = await acreate_progress_artifact(...)

await aupdate_progress_artifact(artifact_id=progress_id, progress=...)

from prefect import flow
from prefect.artifacts import acreate_progress_artifact, aupdate_progress_artifact
import asyncio

@flow
async def progress_ui_test():
    aid = await acreate_progress_artifact(progress=0, key="ui-progress-test", description="UI progress test")
    for i in range(0, 101, 10):
        await aupdate_progress_artifact(artifact_id=aid, progress=i, description=f"{i}%")
        await asyncio.sleep(2)  # allow the UI to poll between updates

progress_ui_test()

ui-progress-test

import time
  last_sent = 0.0

  # inside your async update block
  now = time.monotonic()
  if (now - last_sent) >= 2.0 or downloaded_urls == len(urls):
      last_sent = now
      await aupdate_progress_artifact(
          artifact_id=progress_artifact_id,
          progress=(downloaded_urls / len(urls)) * 100
      )

prefect version

# by artifact id
curl <http://127.0.0.1:4200/api/artifacts/<ARTIFACT_ID>>

# by key (list latest first)
curl "<http://127.0.0.1:4200/api/artifacts?key=frentes-membros-progress&sort=desc>"

prefect version

from prefect import flow
from prefect.futures import resolve_futures_to_results
from prefect.artifacts import create_table_artifact

@flow
def pipeline(...):
    # submit tasks
    tse_fs = [
        extract_tse.with_options(refresh_cache=refresh_cache).submit(name, url)
        for name, url in TSE_ENDPOINTS.items()
    ]

    # resolve results to paths (or whatever you want in the table)
    results = resolve_futures_to_results(tse_fs)

    # build a table: list[dict] works nicely
    table = [
        {"name": name, "url": url, "path": path}
        for (name, url), path in zip(TSE_ENDPOINTS.items(), results)
    ]

    # create/update the aggregate table artifact (new version each run)
    create_table_artifact(
        table=table,
        key="tse-downloads-summary",
        description="Resumo dos downloads TSE desta execução"
    )

    return results

update_progress_artifact

aupdate_progress_artifact

Artifact.get(key="tse-downloads-summary")

from prefect.artifacts import create_table_artifact
from time import monotonic

last_pub = 0.0
def maybe_publish(partial_rows):
    global last_pub
    now = monotonic()
    if now - last_pub >= 5.0 or len(partial_rows) == TOTAL:
        create_table_artifact(
            table=partial_rows,
            key="tse-downloads-summary",
            description="Resumo parcial dos downloads TSE"
        )
        last_pub = now

from prefect.artifacts import acreate_table_artifact

# ...
artifact_id = await acreate_table_artifact(
    key="frentes_membros",
    table=artifact_data,
    description="Total de membros encontrados nas frentes."
)

from prefect.artifacts import create_table_artifact

# ...
artifact_id = await create_table_artifact(
    key="frentes_membros",
    table=artifact_data,
    description="Total de membros encontrados nas frentes."
)

deputados_f = extract_deputados.submit(legislatura)
  assiduidade_f = extract_assiduidade_deputados.submit(deputados_f, legislatura)

from prefect import flow, task
from prefect.futures import resolve_futures_to_results

@task
def extract_assiduidade_for_one(dep_id: str, legislatura: int) -> dict:
    # ... compute one deputado’s assiduidade ...
    return {"id": dep_id, "ok": True}

@flow
def extract_assiduidade_deputados(ids: list[str], legislatura: int) -> list[dict]:
    fs = [extract_assiduidade_for_one.submit(i, legislatura) for i in ids]
    return resolve_futures_to_results(fs)

# In your main flow:
deputados_f = extract_deputados.submit(legislatura)
assiduidade_f = extract_assiduidade_deputados.submit(deputados_f, legislatura)

deputados_ids = await deputados_f.result()  # or just await deputados_f if you prefer
assiduidade_fs = [extract_assiduidade_for_one.submit(i, legislatura) for i in deputados_ids]

.submit(deputados_f, legislatura)

.submit

.result()

deputados_f

extract_assiduidade_deputados

from pathlib import Path
from typing import Sequence, cast
from uuid import UUID
from prefect import task, get_run_logger
from prefect.artifacts import (
    acreate_progress_artifact,
    aupdate_progress_artifact,
)
from selectolax.parser import HTMLParser
import re

from <http://utils.io|utils.io> import fetch_html_many_async, save_ndjson
from config.loader import load_config

APP_SETTINGS = load_config()

def assiduidade_urls(deputados_ids: Sequence[str], leg_years: Sequence[int]) -> list[str]:
    return [
        f"{APP_SETTINGS.CAMARA.PORTAL_BASE_URL}deputados/{dep_id}/presenca-plenario/{year}"
        for dep_id in deputados_ids
        for year in leg_years
    ]

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=APP_SETTINGS.CAMARA.TIMEOUT,
)
async def extract_assiduidade_deputados(
    deputados_ids: list[str],
    legislatura: dict,
    out_dir: str | Path = "data/camara/assiduidade",
) -> str:
    logger = get_run_logger()

    # Get years safely
    leg_years = legislatura.get("dados", {}).get("anosPassados", [])
    if not isinstance(leg_years, list):
        leg_years = []

    # Normalize IDs to strings
    dep_ids = [str(i) for i in deputados_ids]

    progress_id = await acreate_progress_artifact(
        progress=0.0,
        description="Progresso do download da assiduidade de deputados",
    )
    <http://logger.info|logger.info>(f"Progress artifact id: {progress_id}")

    urls = assiduidade_urls(dep_ids, cast(list[int], leg_years))
    <http://logger.info|logger.info>(f"Câmara: buscando assiduidade de {len(dep_ids)} deputados dos anos {leg_years}")

    htmls = await fetch_html_many_async(
        urls=urls,
        concurrency=APP_SETTINGS.CAMARA.CONCURRENCY,
        timeout=APP_SETTINGS.CAMARA.TIMEOUT,
        progress_artifact_id=progress_id,
    )

    await aupdate_progress_artifact(
        artifact_id=progress_id,
        progress=100.0,
        description="Downloads concluídos",
    )

    href_pattern = re.compile(r"<https://www>\.camara\.leg\.br/deputados/\d+")
    id_ano_pattern = r"/deputados/(?P<id>\d+)\?.*ano=(?P<ano>\d+)"

    json_results = []
    for html in htmls:
        tree = HTMLParser(cast(str, html))
        for link in tree.css("a"):
            href = link.attributes.get("href", "") or ""
            if not isinstance(href, str):
                logger.warning(f"O href {href} não é string")
                continue
            if href_pattern.match(href):
                match = re.search(id_ano_pattern, href)
                if match:
                    deputado_id = int(match.group("id"))
                    ano = int(match.group("ano"))
                    json_results.append({"deputado_id": deputado_id, "ano": ano, "html": html})
                else:
                    logger.warning("Não foram encontrados dados suficientes na página HTML")

    dest = Path(out_dir) / "assiduidade.ndjson"
    dest_path = save_ndjson(json_results, dest)
    return dest_path

extract_deputados

list[str]

PrefectFuture[list[str]]

extract_deputados

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=APP_SETTINGS.CAMARA.TASK_TIMEOUT_SECONDS,  # e.g., 900
)

htmls = await fetch_html_many_async(
    urls=urls,
    concurrency=APP_SETTINGS.CAMARA.CONCURRENCY,
    timeout=APP_SETTINGS.CAMARA.HTTP_TIMEOUT_SECONDS,  # e.g., 20–60
    progress_artifact_id=progress_id,
)

assiduidade_f = extract_assiduidade_deputados.with_options(
    timeout_seconds=900  # or None to disable
).submit(deputados_f, legislatura)

# safer:
  leg_years = (legislatura.get("dados") or {}).get("anosPassados") or []
  # ensure ints
  leg_years = [int(y) for y in leg_years]

from datetime import date
from prefect import task
from prefect.utilities.hashing import hash_objects

def assiduidade_cache_key(context, params) -> str | None:
    # Expect params to include legislatura (dict) and deputados_ids (list[str])
    today_year = date.today().year
    leg_years = (params.get("legislatura") or {}).get("dados", {}).get("anosPassados") or []
    dep_ids = params.get("deputados_ids") or []

    # If any requested year is current year, skip caching this run entirely
    # (You could be more granular if you split per-year work into separate tasks)
    if any(int(y) == today_year for y in leg_years):
        return None

    # Otherwise, cache by the combination of dep_ids and leg_years
    # Normalize to strings and a stable ordering
    key = hash_objects(sorted(map(str, dep_ids)), sorted(int(y) for y in leg_years))
    return f"assiduidade:{key}"

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=1800,
    cache_key_fn=assiduidade_cache_key,
    cache_expiration=timedelta(days=30),  # adjust to taste
)
async def extract_assiduidade_deputados(...):
    ...

from datetime import date, timedelta
from prefect.cache_policies import CachePolicy
from prefect.utilities.hashing import hash_objects

def assiduidade_cache_key(context, params) -> str | None:
    today_year = date.today().year
    leg_years = (params.get("legislatura") or {}).get("dados", {}).get("anosPassados") or []
    dep_ids = params.get("deputados_ids") or []
    if any(int(y) == today_year for y in leg_years):
        return None
    key = hash_objects(sorted(map(str, dep_ids)), sorted(int(y) for y in leg_years))
    return f"assiduidade:{key}"

ASSIDUIDADE_CACHE = CachePolicy.from_cache_key_fn(
    assiduidade_cache_key,
    # optional: key_storage=..., isolation_level=..., lock_manager=...
)

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=1800,
    cache_policy=ASSIDUIDADE_CACHE,
    cache_expiration=timedelta(days=30),
)
async def extract_assiduidade_deputados(...):
    ...

@task(cache_key_fn=lambda _, p: None if p["year"] == date.today().year else f"assiduidade:{p['year']}", cache_expiration=timedelta(days=180))
async def extract_assiduidade_year(deputados_ids: list[str], year: int) -> str:
    # do the same logic but for one year
    ...

# In your flow after you have dep_ids and leg_years:
fs = [extract_assiduidade_year.submit(dep_ids, y) for y in leg_years]

leg_years = (legislatura.get("dados") or {}).get("anosPassados") or []
  leg_years = [int(y) for y in leg_years]

[0]

dados

cache_key_fn=lambda _, year: None if year == date.today().year else f"assiduidade:{year}"

"assiduidade:{'deputados_ids': [...], 'legislatura_year': 2022, 'out_dir': '...'}"

from datetime import date, timedelta
from prefect.utilities.hashing import hash_objects

def assiduidade_cache_key(_, params) -> str | None:
    # pull year from params dict
    year = int(params["legislatura_year"])
    # skip caching for the current year
    if year == date.today().year:
        return None

    # Include deputies list in the key so cache changes if inputs change
    dep_ids = [str(i) for i in params.get("deputados_ids", [])]
    deps_hash = hash_objects(sorted(dep_ids)) or "noids"

    # Return a short, filesystem-safe key (avoid :, {}, [], spaces, etc.)
    return f"assiduidade-{year}-{deps_hash}"

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=1800,
    cache_key_fn=assiduidade_cache_key,
    cache_expiration=timedelta(days=180),
)
async def extract_assiduidade_deputados(
    deputados_ids: list[str],
    legislatura_year: int,
    out_dir: str | Path = "data/camara",
) -> str:
    ...

@task(
    ...,
    cache_key_fn=lambda _, p: None if int(p["legislatura_year"]) == date.today().year
                              else f"assiduidade-{int(p['legislatura_year'])}",
    cache_expiration=timedelta(days=180)
)

import os
from pathlib import Path
import json

def save_ndjson(records: list[dict], dest_path: str | Path) -> str:
    dest_path = Path(dest_path)
    dest_path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = dest_path.with_suffix(dest_path.suffix + ".tmp")

    with open(tmp_path, "w", encoding="utf-8") as f:
        for rec in records:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")

    # Atomic replace on the same filesystem
    os.replace(tmp_path, dest_path)
    return str(dest_path)

# inside extract_assiduidade_deputados(...)
dest = Path(out_dir) / f"assiduidade_{legislatura_year}.ndjson"
dest_path = save_ndjson(json_results, dest)
return dest_path

import os, shutil
from pathlib import Path

def merge_ndjson(inputs: list[str | Path], dest: str | Path) -> str:
    dest = Path(dest)
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_suffix(dest.suffix + ".tmp")

    with open(tmp, "w", encoding="utf-8") as out:
        for p in inputs:
            p = Path(p)
            if not p.exists():
                continue
            with open(p, "r", encoding="utf-8") as f:
                shutil.copyfileobj(f, out)

    os.replace(tmp, dest)
    return str(dest)

# submit one task per year (what you’re already doing)
assiduidade_fs = [
    extract_assiduidade_deputados.submit(cast(Any, deputados_f), ano)
    for ano in anos_passados
]

# later, merge all NDJSONs deterministically
from prefect.futures import resolve_futures_to_results
paths = resolve_futures_to_results(assiduidade_fs)
final_path = merge_ndjson(paths, Path("data/camara") / "assiduidade.ndjson")

# create a single aggregate artifact here if desired
from prefect.artifacts import create_table_artifact
# build an aggregated table across years, then:
# create_table_artifact(key="assiduidade", table=agg_rows, description="...")

from filelock import FileLock

def append_ndjson(records: list[dict], dest_path: str | Path):
    dest_path = Path(dest_path)
    dest_path.parent.mkdir(parents=True, exist_ok=True)
    lock = FileLock(str(dest_path) + ".lock")
    with lock:
        with open(dest_path, "a", encoding="utf-8") as f:
            for rec in records:
                f.write(json.dumps(rec, ensure_ascii=False) + "\n")

import os, json
from pathlib import Path
from uuid import uuid4

def save_ndjson(records: list[dict], dest_path: str | Path) -> str:
    dest_path = Path(dest_path)
    dest_path.parent.mkdir(parents=True, exist_ok=True)
    # unique tmp file in the same directory to keep replace atomic
    tmp_path = dest_path.with_suffix(dest_path.suffix + f".{os.getpid()}.{uuid4().hex}.tmp")

    try:
        with open(tmp_path, "w", encoding="utf-8") as f:
            for rec in records:
                f.write(json.dumps(rec, ensure_ascii=False) + "\n")
        os.replace(tmp_path, dest_path)  # atomic swap on same filesystem
    finally:
        # best-effort cleanup if something went wrong before replace
        if tmp_path.exists():
            try:
                tmp_path.unlink()
            except OSError:
                pass

    return str(dest_path)

from datetime import date, timedelta

def assiduidade_cache_key(_, p) -> str | None:
    year = int(p["legislatura_year"])
    if year == date.today().year:
        return None  # skip cache for current year
    return f"assiduidade-{year}"  # short, filesystem-safe key

@task(
    retries=APP_SETTINGS.CAMARA.RETRIES,
    retry_delay_seconds=APP_SETTINGS.CAMARA.RETRY_DELAY,
    timeout_seconds=1800,
    cache_key_fn=assiduidade_cache_key,
    cache_expiration=timedelta(days=180),
)
async def extract_assiduidade_deputados(...):
    ...

dest = Path(out_dir) / f"assiduidade_{legislatura_year}.ndjson"
dest_path = save_ndjson(json_results, dest)
return dest_path

# submit assiduidade tasks
assiduidade_fs = [
    extract_assiduidade_deputados.submit(cast(Any, deputados_f), ano)
    for ano in anos_passados
]

# submit independent branch BEFORE waiting
frentes_f = extract_frentes.submit(legislatura)
frentes_membros_f = extract_frentes_membros.submit(cast(Any, frentes_f))

# now resolve and merge
paths = resolve_futures_to_results(assiduidade_fs)
final_path = merge_ndjson(paths, Path("data/camara") / "assiduidade.ndjson")

@task
def merge_ndjson_task(paths: list[str], dest: str | Path) -> str:
    return merge_ndjson(paths, dest)

merged_f = merge_ndjson_task.submit(assiduidade_fs, Path("data/camara") / "assiduidade.ndjson")

await acreate_table_artifact(
    key=f"assiduidade-{legislatura_year}",
    table=artifact_data,
    description=f"Assiduidade de deputados ({legislatura_year})",
)

dados = (legislatura.get("dados") or {})
anos_passados = dados.get("anosPassados", [])

extract_assiduidade_deputados.with_options(refresh_cache=refresh_cache).submit(...)

import os, json
from pathlib import Path
from uuid import uuid4

def save_ndjson(records: list[dict], dest_path: str | Path) -> str:
    dest_path = Path(dest_path)
    dest_path.parent.mkdir(parents=True, exist_ok=True)
    # unique tmp in same directory (same filesystem) to keep replace atomic
    tmp_path = dest_path.with_name(dest_path.name + f".{os.getpid()}.{uuid4().hex}.tmp")

    try:
        with open(tmp_path, "w", encoding="utf-8") as f:
            for rec in records:
                f.write(json.dumps(rec, ensure_ascii=False) + "\n")
        os.replace(tmp_path, dest_path)  # atomic swap on same volume
    finally:
        # best-effort cleanup if something went wrong before replace
        if tmp_path.exists():
            try:
                tmp_path.unlink()
            except OSError:
                pass

    return str(dest_path)

anos_passados = sorted(set(anos_passados))

assiduidade_{year}_{flow_run_id}.ndjson

assiduidade_{year}.ndjson

assert Path(tmp_path).exists()

import os, json
from pathlib import Path

def save_ndjson(records: list[dict], dest_path: str | Path) -> str:
    dest_path = Path(dest_path)
    dest_path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = dest_path.with_suffix(dest_path.suffix + ".tmp")

    with open(tmp_path, "w", encoding="utf-8") as f:
        for rec in records:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")

    os.replace(tmp_path, dest_path)  # atomic on same filesystem
    return str(dest_path)

import os, json
from pathlib import Path
from uuid import uuid4

def save_ndjson(records: list[dict], dest_path: str | Path) -> str:
    dest_path = Path(dest_path)
    dest_path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = dest_path.with_name(dest_path.name + f".{os.getpid()}.{uuid4().hex}.tmp")

    try:
        with open(tmp_path, "w", encoding="utf-8") as f:
            for rec in records:
                f.write(json.dumps(rec, ensure_ascii=False) + "\n")
        os.replace(tmp_path, dest_path)
    finally:
        if tmp_path.exists():
            try:
                tmp_path.unlink()
            except OSError:
                pass
    return str(dest_path)