prefect config set PREFECT_API_URL="<http://127.0.0.1:4200/api>"

systemctl enable prefect-worker

systemctl start prefect-worker

--help

prefect deploy

$ prefect init

python
import os

# Inside your flow or task
script_path = os.path.abspath(__file__)
print(f"The flow is running from: {script_path}")

__file__

os.path.abspath

prefect.yml

prefect.yml

prefect.yml

prefect.yml

prefect.yml

prefect.yml

prefect.yml

prefect.yml

prefect deploy

@flow

__main__

python
@flow
def basic_flow(message="Hola mundo!"):
    print_message(message)

if __name__ == "__main__":
    basic_flow()

shell
> prefect deploy
? Select a flow to deploy [Use arrows to move; enter to select; n to select none]
┏━━━┳━━━━━━━━━━━━┳━━━━━━━━━━┓
┃   ┃ Flow Name  ┃ Location ┃
┡━━━╇━━━━━━━━━━━━╇━━━━━━━━━━┩
│ > │ basic_flow │ basic.py │
└───┴────────────┴──────────┘
    Enter a flow entrypoint manually

shell
? Flow entrypoint (expected format path/to/file.py:function_name): ./project/basic.py:basic_flow

(30 1,13,17 * * *)

compras-prod

shell
? Deployment name (default): printear-mensaje # Ingreso un nombre para el deploy.
? Would you like to configure a schedule for this deployment? [y/n] (y): n # No configuro la ejecución automática

Help me write a simple prefect documentation in markdown in spanish. So far I've written this:

prefect deploy

@flow

__main__

python
@flow
def basic_flow(message="Hola mundo!"):
    print_message(message)

if __name__ == "__main__":
    basic_flow()

prefect deploy

logging

DEBUG

INFO

WARNING

ERROR

CRITICAL

logger.debug()

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

logger.warning()

logger.error()

logger.critical()

logging

logger

python
from prefect import task, Flow
import logging

# Obtiene una instancia del logger de Prefect
logger = logging.getLogger("prefect")

@task
def tarea1():
    # Registra un mensaje informativo
    <http://logger.info|logger.info>("Ejecutando la tarea 1")

@task
def tarea2(x):
    # Registra un mensaje de debug
    logger.debug(f"Tarea 2 recibió el argumento: {x}")

    # Simula alguna validación y registra una advertencia si es necesario
    if x < 0:
        logger.warning("x es menor que 0; esto podría ser un problema.")

    # Regresa algún valor
    return x * 2

# Define un flujo que usa ambas tareas
with Flow("mi_flujo_de_ejemplo") as flow:
    resultado1 = tarea1()
    resultado2 = tarea2(resultado1)

# Ejecuta el flujo
if __name__ == "__main__":
    flow.run()

tarea1

tarea2

python
logging.basicConfig(level=<http://logging.INFO|logging.INFO>)

INFO

INFO

WARNING

ERROR

CRITICAL

DEBUG

logging.DEBUG

logging

logging

python
from prefect import flow, task
import prefect

@task
def mi_tarea():
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>("Ejecutando tarea")

@flow
def mi_flujo():
    mi_tarea()

mi_flujo()

prefect.context.get("logger")

[Código Local] --> |define| --> [Flujo (Flow)]
   |
   |-----|deploy|----> [Archivo prefect.yml] --> |registra en| --> [Servidor Prefect]
   |                                                     |
   |                                                     |--> |crea| --> [Deployment]
   |                                                               |
   |                                                               |--> |asigna| --> [Work Pool]
   |                                                                                 |
   |                                                                                 |--> [Workers]
   |                                                                                                |
   |                                                                                                |--> |escucha| --> [Flow Runs]
                                                                                                                  |
                                                                                                                  |--> |ejecuta| --> [Tareas (Tasks)]

prefect.yml

mermaid
graph TD
    Código_Local["Código Local"] --> |define| Flujo["Flujo (Flow)"]
    Código_Local --> |deploy| Prefect_yml["Archivo prefect.yml"]
    Prefect_yml --> |registra en| Servidor_Prefect["Servidor Prefect"]

    Servidor_Prefect --> |crea| Deployment["Deployment"]
    Deployment --> |asigna| Work_Pool["Work Pool"]
    Work_Pool --> Workers["Workers"]
    Workers --> |escucha| Flow_Runs["Flow Runs"]
    Flow_Runs --> |ejecuta| Tareas["Tareas (Tasks)"]

mermaid
graph TB
    Código_Local["Código Local"] --> |define| Flujo
    Código_Local --> |deploy| Prefect_yml["Archivo prefect.yml"]
    Prefect_yml --> |registra en| Servidor_Prefect["Servidor Prefect"]:::servidor

    subgraph " "
        direction LR
        Servidor_Prefect --> |crea| Deployment["Deployment"]:::deployment
        Deployment --> |asigna| Work_Pool["Work Pool"]:::workpool
    end

    subgraph " "
        direction TB
        Work_Pool --> Workers["Workers"]:::workers
        Workers --> |escucha| Flow_Runs["Flow Runs"]:::flowruns
        Flow_Runs --> |ejecuta| Tareas["Tareas (Tasks)"]:::tareas
    end

    classDef default fill:#f9f,stroke:#333,stroke-width:2px;
    classDef servidor fill:#bbf,stroke:#333,stroke-width:2px;
    classDef deployment fill:#fbf,stroke:#f66,stroke-width:2px,color:#f66;
    classDef workpool fill:#fbb,stroke:#333,stroke-width:2px;
    classDef workers fill:#bfb,stroke:#333,stroke-width:2px;
    classDef flowruns fill:#ccf,stroke:#333,stroke-width:2px;
    classDef tareas fill:#ffc,stroke:#333,stroke-width:2px;

    class Flujo default;

Escribe en español y de manera resumida con un unico ejemplo esto:

@flow

@flow

prefect work-pool create [OPCIONES] NOMBRE

NOMBRE

python
from prefect import config

# Obtener la URL de la API de Prefect desde la configuración
prefect_api_url = config.get("PREFECT_API_URL")

print(f"La URL de la API de Prefect es: {prefect_api_url}")

config

prefect config set PREFECT_API_URL="tu_url"

<https://orion-docs.prefect.io/migration_guide/>

prefect.config

prefect.settings

prefect.config

python
from prefect import get_settings

# Get the Orion API URL
prefect_api_url = get_settings().api_url

print(f"La URL de la API de Prefect es: {prefect_api_url}")

prefect.settings

PREFECT_API_URL

PREFECT_API_KEY

bash
prefect config set PREFECT_API_URL="<https://api.prefect.cloud/api/accounts/[ACCOUNT-ID]/workspaces/[WORKSPACE-ID]>"
prefect config set PREFECT_API_KEY="[API-KEY]"

python
import os

PREFECT_API_URL = os.getenv("PREFECT_API_URL")
PREFECT_API_KEY = os.getenv("PREFECT_API_KEY")

python
from prefect import Client

client = Client()

flow_run_id = "id-of-the-flow-run"  # Replace with the actual flow run ID
new_tag = "new-tag"

# Add a tag to an existing flow run
client.update_flow_run_tags(flow_run_id=flow_run_id, add=[new_tag])

update_flow_run_tags

sys.path

concurrent

python
import sys

# Add the directories of your flows to the Python path if necessary
sys.path.append('/path/to/flow1_directory')
sys.path.append('/path/to/flow2_directory')

from flow1_directory.flow1 import flow1
from flow2_directory.flow2 import flow2

from prefect import flow

@flow
def parent_flow():
    # Run flows sequentially
    flow1_run = flow1()
    flow2_run = flow2()

    # Alternatively, run flows in parallel using Prefect's concurrent features
    # to be adjusted according to your actual use case

parent_flow()

flow1

flow2

parent_flow

parent_flow

When I run prefect deploy this appears:

env

.venvs

PYTHONPATH

.prefectignore

.gitignore

.prefectignore

plaintext
   env/
   **/__pycache__/
   **/*.pyc

prefect deploy

__file__

Prefect's context

parameter passing

project/
├── flow_script.py
└── some_module/
    └── task_module.py

task_module.py

python
# project/some_module/task_module.py
from prefect import task

@task
def my_task():
    import __main__
    flow_script_path = __main__.__file__
    print(f"The flow script path is: {flow_script_path}")

flow_script.py

python
# project/flow_script.py
from prefect import flow
from some_module.task_module import my_task

@flow
def my_flow():
    my_task()

if __name__ == "__main__":
    my_flow()

flow_script.py

my_task

flow_script.py

__file__

__main__

__file__

flow_script.py

some_module

project

bash
python flow_script.py

plaintext
The flow script path is: project/flow_script.py

__main__

bash
prefect variables set MY_VARIABLE "Some Value"

python
from prefect import variables

# Set a variable
variables.set_variable("MY_VARIABLE", "Some Value")

python
from prefect import flow, task, variables

@task
def read_variable():
    value = variables.get_variable("MY_VARIABLE")
    print(f"The value of MY_VARIABLE is: {value}")
    return value

@flow
def monitor_flow():
    read_variable()

if __name__ == "__main__":
    monitor_flow()

python
from prefect import flow, task, variables

@task
def set_variable():
    variables.set_variable("MY_VARIABLE", "Execution Complete")

@flow
def execution_flow():
    set_variable()

if __name__ == "__main__":
    execution_flow()

python
from prefect import flow, task, variables

@task
def read_variable():
    value = variables.get_variable("MY_VARIABLE")
    print(f"The value of MY_VARIABLE is: {value}")
    return value

@flow
def monitor_flow():
    read_variable()

if __name__ == "__main__":
    monitor_flow()

execution_flow.py

bash
python execution_flow.py

monitoring_flow.py

bash
python monitoring_flow.py

python
from prefect import flow, task, variables

@task
def set_var():
    variables.set_variable("VAR1", "Value from script 1")

@flow
def script1_flow():
    set_var()

if __name__ == "__main__":
    script1_flow()

python
from prefect import flow, task, variables

@task
def set_var():
    variables.set_variable("VAR2", "Value from script 2")

@flow
def script2_flow():
    set_var()

if __name__ == "__main__":
    script2_flow()

python
from prefect import flow, task, variables

@task
def read_variables():
    var1 = variables.get_variable("VAR1")
    var2 = variables.get_variable("VAR2")
    print(f"VAR1: {var1}, VAR2: {var2}")

@flow
def monitoring_flow():
    read_variables()

if __name__ == "__main__":
    monitoring_flow()

python
from prefect import flow, task, context

# Script 1
@task
def set_var():
    context.get_run_context().task_run.state["var_from_script1"] = "Value from script 1"

@flow
def script1_flow():
    set_var()

if __name__ == "__main__":
    script1_flow()

# Script 2
@task
def set_var():
    context.get_run_context().task_run.state["var_from_script2"] = "Value from script 2"

@flow
def script2_flow():
    set_var()

if __name__ == "__main__":
    script2_flow()

# Monitoring Script
@task
def read_variables():
    var1 = context.get_run_context().variables.get("var_from_script1")
    var2 = context.get_run_context().variables.get("var_from_script2")
    print(f"var_from_script1: {var1}, var_from_script2: {var2}")

@flow
def monitoring_flow():
    read_variables()

if __name__ == "__main__":
    monitoring_flow()

python
from prefect import flow, task
from prefect.client import get_client

@task
def set_metadata():
    client = get_client()
    flow_run_id = client.get_current_flow_run().id
    metadata = {"script": "script1", "info": "This is script 1 execution"}
    
    client.update_flow_run_metadata(flow_run_id, metadata)
    return metadata

@flow
def flow1():
    set_metadata()

if __name__ == "__main__":
    flow1()

python
from prefect import flow, task
from prefect.client import get_client

@task
def set_metadata():
    client = get_client()
    flow_run_id = client.get_current_flow_run().id
    metadata = {"script": "script2", "info": "This is script 2 execution"}
    
    client.update_flow_run_metadata(flow_run_id, metadata)
    return metadata

@flow
def flow2():
    set_metadata()

if __name__ == "__main__":
    flow2()

python
from prefect import flow, task
from prefect.client import get_client

@task
def get_metadata(flow_run_id):
    client = get_client()
    flow_run_metadata = client.get_flow_run_metadata(flow_run_id)
    return flow_run_metadata

@flow
def monitoring_flow(flow_runs):
    for run_id in flow_runs:
        metadata = get_metadata(run_id)
        print(f"Metadata for flow run {run_id}: {metadata}")

if __name__ == "__main__":
    # Dummy flow run IDs for the demonstration, replace with actual flow run IDs
    flow_runs = ["flow_run_id_1", "flow_run_id_2"]
    monitoring_flow(flow_runs)

bash
   python flow_script1.py
   python flow_script2.py

bash
   python monitoring_script.py

get_client()

set_metadata

update_flow_run_metadata

get_flow_run_metadata

"flow_run_id_1"

"flow_run_id_2"

python
  from prefect.blocks.system import String

  # Creating a String block
  api_endpoint = String(value="<https://api.example.com/data>")
  api_endpoint.save(name="api-endpoint")

  # Using the String block in a flow
  @flow
  def my_flow():
      endpoint = String.load("api-endpoint").value
      print(f"API Endpoint: {endpoint}")

  if __name__ == "__main__":
      my_flow()

python
  from prefect import flow, task, variables

  @task
  def read_variable():
      config_value = variables.get_variable("config_key")
      print(f"Config Value: {config_value}")

  @flow
  def my_flow():
      read_variable()

  if __name__ == "__main__":
      # Set a variable using CLI or programmatically
      variables.set_variable("config_key", "config_value")
      my_flow()