### Plombery Configuration File Example

Source: https://lucafaggianelli.com/plombery/configuration

Example of a `plombery.config.yaml` file showing how to configure frontend URL, authentication details (client ID, client secret, server metadata URL), and notification settings including pipeline status and channels.

```yaml
frontend_url:https://pipelines.example.com
auth:
  client_id:$GOOGLE_CLIENT_ID
  client_secret:$GOOGLE_CLIENT_SECRET
  server_metadata_url:https://accounts.google.com/.well-known/openid-configuration
notifications:
-pipeline_status:
-failed
channels:
-$GMAIL_ACCOUNT
-$MSTEAMS_WEBHOOK
```

--------------------------------

### Installing Uvicorn

Source: https://lucafaggianelli.com/plombery/create-a-pipeline

This command installs the uvicorn package, which is a necessary ASGI web server for running Plombery applications.

```bash
pip install uvicorn[standard]
```

--------------------------------

### Install Plombery

Source: https://lucafaggianelli.com/plombery/get-started

Command to install the Plombery library using pip after activating the virtual environment. This downloads and installs the package and its dependencies.

```bash
pip install plombery
```

--------------------------------

### Create Virtual Environment

Source: https://lucafaggianelli.com/plombery/get-started

Command to create a Python virtual environment in the project folder. This isolates project dependencies.

```bash
python -m venv .venv
```

--------------------------------

### Running the Plombery Application

Source: https://lucafaggianelli.com/plombery/create-a-pipeline

This code snippet shows how to start the Plombery application using `uvicorn`. It's typically placed in the `if __name__ == "__main__":` block to ensure it runs when the script is executed directly.

```python
if __name__ == "__main__":
    import uvicorn

    uvicorn.run("plombery:get_app", reload=True, factory=True)
```

--------------------------------

### Activate Virtual Environment (Windows)

Source: https://lucafaggianelli.com/plombery/get-started

Command to activate the created virtual environment on Windows systems. This ensures that subsequent commands use the isolated environment's Python interpreter and packages.

```bash
.venv\Script\activate
```

--------------------------------

### Registering a Plombery Pipeline

Source: https://lucafaggianelli.com/plombery/pipelines

Demonstrates how to register a new pipeline using `register_pipeline`. It includes mandatory fields like `id` and `tasks`, and optional fields such as `params`, `name`, `description`, and `triggers`. The example also shows how to define input parameters using Pydantic models and configure interval-based triggers.

```python
from plombery import register_pipeline, task
from pydantic import BaseModel
from plombery.triggers import Trigger, IntervalTrigger

class InputParams(BaseModel):
  some_value: int

@task
def get_sales_data():
  pass

register_pipeline(
    # (required) the id identifies the pipeline univocally
    id="sales_pipeline_2345",
    # (required) the list of tasks to execute
    tasks=[get_sales_data],
    # This pipeline is configurable via input parameters
    params=InputParams,
    # The name is optional, if absent it would be generated from the ID
    name="Sales pipeline",
    description="""This is a very useless pipeline""",
    # Triggers with schedules
    triggers=[
        Trigger(
            id="daily",
            name="Daily",
            description="Run the pipeline every day",
            # the input params value for this specific trigger
            params=InputParams(some_value=2),
            schedule=IntervalTrigger(
                days=1,
            ),
        )
    ],
)
```

--------------------------------

### Activate Virtual Environment (Mac/Linux)

Source: https://lucafaggianelli.com/plombery/get-started

Command to activate the created virtual environment on macOS and Linux systems. This ensures that subsequent commands use the isolated environment's Python interpreter and packages.

```bash
source .venv/bin/activate
```

--------------------------------

### Environment Variables for Secrets

Source: https://lucafaggianelli.com/plombery/configuration

Example of a `.env` file used to store sensitive information such as Google client ID, client secret, and notification channel credentials. Plombery automatically loads `.env` files from the project root.

```env
# Auth
GOOGLE_CLIENT_ID="ABC123"
GOOGLE_CLIENT_SECRET="DEF456"

# Notifications
GMAIL_ACCOUNT=mailto://myuser:mypass@gmail.com
MSTEAMS_WEBHOOK=msteams://TokenA/TokenB/TokenC/
```

--------------------------------

### SSL Certificate Check Pipeline Example

Source: https://lucafaggianelli.com/plombery/recipes/certificates-expiration

This Python code demonstrates a Plombery pipeline designed to check SSL certificate expiration dates for a list of hostnames. It utilizes Plombery's task and pipeline functionalities to automate this process. The pipeline fetches certificate information and can be configured to trigger notifications for upcoming expirations.

```python
from plombery import Pipeline, task

@task
def check_ssl_certificate(hostname: str):
    """Checks the SSL certificate expiration date for a given hostname."""
    import ssl
    import socket
    from datetime import datetime

    try:
        context = ssl.create_default_context()
        with socket.create_connection((hostname, 443)) as sock:
            with context.wrap_socket(sock, server_hostname=hostname) as ssock:
                cert = ssock.getpeercert()
                expiry_date = datetime.strptime(cert['notAfter'], '%b %d %H:%M:%S %Y %Z')
                return {
                    "hostname": hostname,
                    "expires_on": expiry_date.isoformat(),
                    "is_valid": expiry_date > datetime.now()
                }
    except Exception as e:
        return {
            "hostname": hostname,
            "error": str(e)
        }

@task
def notify_if_expiring(certificate_info: dict):
    """Notifies if a certificate is expiring soon."""
    if certificate_info.get("is_valid") == False:
        print(f"ALERT: Certificate for {certificate_info['hostname']} has expired!")
    elif certificate_info.get("expires_on"):
        expires_on = datetime.fromisoformat(certificate_info['expires_on'])
        if (expires_on - datetime.now()).days < 30:
            print(f"WARNING: Certificate for {certificate_info['hostname']} expires on {certificate_info['expires_on']}")


ssl_pipeline = Pipeline([
    (check_ssl_certificate, {"hostname": "google.com"}),
    (check_ssl_certificate, {"hostname": "github.com"}),
    (check_ssl_certificate, {"hostname": "expired.badssl.com"}),
])

# To run the pipeline:
# ssl_pipeline.run()

# To run with notification:
# for cert_info in ssl_pipeline.run():
#     notify_if_expiring(cert_info)

```

--------------------------------

### Basic Plombery Project Structure

Source: https://lucafaggianelli.com/plombery/create-a-pipeline

Demonstrates the essential file and folder structure for a new Plombery project. It includes a virtual environment, an empty `__init__.py` file to declare the directory as a Python package, and the main application file (`app.py`).

```bash
.
├── venv/
├── __init__.py
└── app.py
```

--------------------------------

### Running Plombery with Uvicorn

Source: https://lucafaggianelli.com/plombery/create-a-pipeline

This Python code shows how to run the Plombery application using the uvicorn server. It's typically used within an `if __name__ == "__main__":` block for direct execution.

```python
if __name__ == "__main__":
    import uvicorn
    uvicorn.run("plombery:get_app", reload=True, factory=True)
```

--------------------------------

### Plombery System Configuration

Source: https://lucafaggianelli.com/plombery/configuration

Configures essential system parameters for Plombery, including database connection, allowed origins for cross-origin requests, and the frontend URL.

```APIDOC
System Configuration:
  database_url: The connection URL for the Plombery database.
  allowed_origins: A list of allowed origins for cross-origin requests.
  frontend_url: The URL of the Plombery frontend application.
```

--------------------------------

### Registering a Sales Pipeline

Source: https://lucafaggianelli.com/plombery/create-a-pipeline

This Python code snippet demonstrates how to register a pipeline named 'sales_pipeline'. It includes a description, a task 'fetch_raw_sales_data', and a daily trigger.

```python
def fetch_raw_sales_data():
    # Placeholder for fetching sales data
    sales = [...] # Assume sales data is fetched here
    return sales

register_pipeline(
    id="sales_pipeline",
    description="Aggregate sales activity from all stores across the country",
    tasks=[fetch_raw_sales_data],
    triggers=[
        Trigger(
            id="daily",
            name="Daily",
            description="Run the pipeline every day",
            schedule=IntervalTrigger(days=1),
        ),
    ],
)
```

--------------------------------

### Defining a Plombery Task for Sales Data

Source: https://lucafaggianelli.com/plombery/create-a-pipeline

This Python code defines an asynchronous task using the `@task` decorator. The `fetch_raw_sales_data` function simulates fetching sales data, logs its progress using Plombery's logger, and generates random sales records.

```python
from datetime import datetime
from random import randint

from apscheduler.triggers.interval import IntervalTrigger
from plombery import task, get_logger, Trigger, register_pipeline


@task
async def fetch_raw_sales_data():
"""Fetch latest 50 sales of the day"""

    # using Plombery logger your logs will be stored
    # and accessible on the web UI
    logger = get_logger()

    logger.debug("Fetching sales data...")

    sales = [
        {
            "price": randint(1, 1000),
            "store_id": randint(1, 10),
            "date": datetime.today(),
            "sku": randint(1, 50),
        }
        for _ in range(50)
    ]

    logger.info("Fetched %s sales data rows", len(sales))
```

--------------------------------

### Authentication Settings

Source: https://lucafaggianelli.com/plombery/configuration

Configuration for Plombery's authentication system, which is based on OAuth providers and Authlib. It includes client credentials, server metadata URLs, and optional client arguments like scope.

```yaml
auth:
  client_id: "your_client_id"
  client_secret: "your_client_secret"
  server_metadata_url: "https://accounts.google.com/.well-known/openid-configuration"
  # or specify individual URLs:
  # access_token_url: "..."
  # authorize_url: "..."
  # jwks_uri: "..."
  client_kwargs:
    scope: "openid email profile"
  secret_key: "a_very_secret_key_for_production"
```

--------------------------------

### Register Pipeline with Input Parameters

Source: https://lucafaggianelli.com/plombery/pipelines

Demonstrates how to register a Plombery pipeline with configurable input parameters using the `params` argument and a Pydantic `BaseModel`.

```python
register_pipeline(
  # ...
  params=InputParams
)
```

--------------------------------

### Plombery Features Overview

Source: https://lucafaggianelli.com/plombery/index

This section outlines the core features of Plombery, highlighting its capabilities for task scheduling, web interface, pipeline definition, parametrization, security, and monitoring.

```APIDOC
Plombery Features:
  - Task scheduling using APScheduler (Interval, Cron, Date triggers)
  - Built-in Web interface
  - Pipelines and tasks defined in pure Python
  - Parametrization via Pydantic
  - Manual pipeline execution from Web UI
  - OAuth2 security
  - Debugging with logs and output data
  - Monitoring and alerting
  - REST API for integrations
```

--------------------------------

### Create a Plombery Task

Source: https://lucafaggianelli.com/plombery/create-a-pipeline

Defines a Plombery task using the `@task` decorator. This task simulates fetching sales data, logs the operation using Plombery's logger, and returns the generated data. The output can be used by subsequent tasks or stored.

```python
from datetime import datetime
from random import randint
from plombery import task, get_logger

@task
async def fetch_raw_sales_data():
    """Fetch latest 50 sales of the day"""
    logger = get_logger()
    logger.debug("Fetching sales data...")

    sales = [
        {
            "price": randint(1, 1000),
            "store_id": randint(1, 10),
            "date": datetime.today(),
            "sku": randint(1, 50),
        }
        for _ in range(50)
    ]

    logger.info("Fetched %s sales data rows", len(sales))
    return sales
```

--------------------------------

### Plombery Authentication Configuration

Source: https://lucafaggianelli.com/plombery/configuration

Sets up authentication mechanisms for Plombery, including client credentials for OAuth 2.0, metadata URLs, token endpoints, and JWKS URI for token validation.

```APIDOC
AuthSettings:
  client_id: The client ID for OAuth 2.0 authentication.
  client_secret: The client secret for OAuth 2.0 authentication.
  server_metadata_url: The URL to fetch OAuth 2.0 server metadata.
  access_token_url: The URL to obtain an access token.
  authorize_url: The URL to authorize the client.
  jwks_uri: The URI to fetch JSON Web Key Set (JWKS) for token verification.
  client_kwargs: Additional keyword arguments for the OAuth client.
  secret_key: A secret key used for session management or other security purposes.
```

--------------------------------

### Registering a Plombery Pipeline

Source: https://lucafaggianelli.com/plombery/create-a-pipeline

This snippet demonstrates how to register a pipeline with a unique ID, description, tasks, and triggers. It utilizes the `register_pipeline` function and defines a daily trigger using `IntervalTrigger`.

```python
register_pipeline(
    id="sales_pipeline",
    description="Aggregate sales activity from all stores across the country",
    tasks = [fetch_raw_sales_data],
    triggers = [
        Trigger(
            id="daily",
            name="Daily",
            description="Run the pipeline every day",
            schedule=IntervalTrigger(days=1),
        ),
    ],
)
```

--------------------------------

### Define Pipeline with Parameterized Triggers

Source: https://lucafaggianelli.com/plombery/triggers

Demonstrates how to define input parameters using Pydantic, register a pipeline with these parameters, and associate triggers with specific parameter values. This allows for alternative entrypoints and scheduled runs with custom configurations.

```python
from pydantic import BaseModel

class InputParams(BaseModel):
    past_days: int
    convert_currency: bool = False

def get_sales_data(params: InputParams):
    print(params.past_days)

register_pipeline(
    id="sales_pipeline",
    tasks=[get_sales_data],
    params=InputParams,
    triggers=[
        Trigger(
            id="daily",
            description="Get last 5 days of sales data in USD dollars",
            schedule=IntervalTrigger(
                days=1,
            ),
            params={
                "past_days": 5,
                "convert_currency": True,
            },
        ),
    ],
)
```

--------------------------------

### Defining Input Parameters for Pipelines

Source: https://lucafaggianelli.com/plombery/tasks

Illustrates how to declare input parameters for a Plombery pipeline using Pydantic's `BaseModel`. These parameters are then passed to the tasks within the pipeline.

```python
class InputParams(BaseModel):
  some_value: int

register_pipeline(
  # ...
  params=InputParams
)
```

--------------------------------

### Notification Rules Configuration

Source: https://lucafaggianelli.com/plombery/configuration

Defines rules for sending notifications based on pipeline run status and specifies notification channels. Supports various pipeline statuses like 'completed', 'failed', and 'cancelled', and multiple channel types including email and MS Teams via Apprise URIs.

```yaml
notifications:
  # Send notifications only if the pipelines failed
  - pipeline_status:
      - failed
    channels:
      # Send them to my gmail address (from my address itself)
      # Better to use an env var here
      - mailto://myuser:mypass@gmail.com
  # Send notifications only if the pipelines succeeded or was cancelled
  - pipeline_status:
      - completed
      - cancelled
    channels:
      # Send them to a MS Teams channel
      # Better to use an env var here
      - msteams://mychanneltoken
```

--------------------------------

### AuthSettings Schema

Source: https://lucafaggianelli.com/plombery/configuration

Defines the parameters for AuthSettings, including client_id, client_secret, server_metadata_url (or individual URLs like access_token_url, authorize_url, jwks_uri), client_kwargs for additional OAuth parameters, and a secret_key for backend middleware.

```APIDOC
AuthSettings:
  client_id: str
    # An OAuth app client ID
  client_secret: str
    # An OAuth app client secret
  server_metadata_url: str (optional)
    # URL containing OAuth provider specific endpoints. If not provided, access_token_url, authorize_url, and jwks_uri must be provided.
    # Example for Google: https://accounts.google.com/.well-known/openid-configuration
  access_token_url: str (optional)
    # OAuth access token URL
  authorize_url: str (optional)
    # OAuth authorization URL
  jwks_uri: str (optional)
    # OAuth JWKS URI
  client_kwargs: dict (optional)
    # Additional values to pass to the OAuth client during the auth process, e.g., scope.
    # Example: {'scope': 'openid email profile'}
  secret_key: str
    # Secret key used in the backend middleware. Should be a strong, unique value in production.
```

--------------------------------

### Registering Tasks in a Pipeline

Source: https://lucafaggianelli.com/plombery/tasks

Shows how to register defined tasks (both sync and async) with the `register_pipeline` function. This function is used to build and manage the execution flow of a Plombery pipeline.

```python
register_pipeline(
  tasks=[sync_task, async_task]
)
```

--------------------------------

### Plombery Task Output Data Passing

Source: https://lucafaggianelli.com/plombery/tasks

Demonstrates how tasks in a Plombery pipeline pass their return values (output data) as positional arguments to subsequent tasks. Task 1 returns 1, Task 2 uses this value, and Task 3 uses values from both Task 1 and Task 2.

```python
@task
def task_1():
  return 1

@task
def task_2(from_1):
  # from_1 = 1
  return from_1 + 1

@task
def task_3(from_1, from_2):
  # from_1 = 1
  # from_2 = 2
  return from_1 + from_2
```

--------------------------------

### Plombery Notification Configuration

Source: https://lucafaggianelli.com/plombery/configuration

Defines rules for sending notifications, specifying the pipeline status to monitor and the channels through which notifications should be sent.

```APIDOC
NotificationRule:
  pipeline_status: The status of the pipeline that triggers a notification (e.g., 'success', 'failure').
  channels: A list of notification channels to use (e.g., 'email', 'slack').
```

--------------------------------

### NotificationRule Schema

Source: https://lucafaggianelli.com/plombery/configuration

Describes the structure of a NotificationRule, including the pipeline_status and channels parameters. pipeline_status accepts 'completed', 'failed', or 'cancelled'. channels accept Apprise URI strings for various notification providers.

```APIDOC
NotificationRule:
  pipeline_status: list[str]
    # A list of 1 or more pipeline run status among: completed, failed, cancelled
  channels: list[str]
    # A list of 1 or more recipients where to send the notifications.
    # A channel is an Apprise URI string.
    # Examples:
    # Email: mailto://myuser:mypass@gmail.com
    # MS Teams: msteams://TokenA/TokenB/TokenC/
    # AWS SES: ses://user@domain/AccessKeyID/AccessSecretKey/RegionName/email1/
```

--------------------------------

### Valid Trigger Parameter Configuration with Optional Values

Source: https://lucafaggianelli.com/plombery/triggers

Shows a valid trigger parameter configuration where an optional parameter (`past_days`) is provided, demonstrating how to correctly set parameters for triggers, including omitting optional ones if not needed.

```python
Trigger(
    # ...
    params={
        "past_days": 3,
    }
)
```

--------------------------------

### Define Input Parameters with Pydantic BaseModel

Source: https://lucafaggianelli.com/plombery/pipelines

Shows the definition of a Pydantic `BaseModel` to specify the input parameters for a Plombery pipeline. This model dictates the structure and types of the input data.

```python
class InputParams(BaseModel):
  some_value: int
```

--------------------------------

### Manual Pipeline Trigger with Parameters

Source: https://lucafaggianelli.com/plombery/triggers

The manual trigger, accessible via the web UI, allows users to run pipelines. If a pipeline has input parameters defined using Pydantic's BaseModel, a form is automatically generated to customize these parameters before execution.

```APIDOC
Manual Pipeline Run:
  UI Element: Button on pipeline home/page
  Functionality: Initiates a pipeline run.
  Parameter Handling:
    - If pipeline has parameters (defined via Pydantic BaseModel):
      - A dialog presents a form to input/customize parameters.
    - If pipeline has no parameters:
      - Pipeline runs immediately upon clicking the button.
```

--------------------------------

### Define Synchronous and Asynchronous Tasks

Source: https://lucafaggianelli.com/plombery/tasks

Demonstrates how to define both synchronous and asynchronous tasks using the `@task` decorator in Plombery. These functions form the building blocks of Plombery pipelines.

```python
@task
def sync_task():
  pass

@task
async def async_task():
  pass
```

--------------------------------

### HTTP Trigger for Pipeline Execution

Source: https://lucafaggianelli.com/plombery/triggers

The HTTP trigger allows pipelines to be executed via an HTTP POST request. The URL for this trigger can be found on the pipeline's page.

```APIDOC
Pipeline Trigger:
  Method: POST
  URL: <pipeline_url>
  Description: Triggers the pipeline execution via an HTTP POST request.
  Parameters:
    - (Optional) JSON body: Key-value pairs representing pipeline input parameters.
  Returns:
    - Status: 200 OK on successful trigger.
    - Body: Details about the pipeline run.
```

--------------------------------

### Accessing Input Parameters in Tasks

Source: https://lucafaggianelli.com/plombery/tasks

Explains how tasks receive the defined input parameters. The `params` argument in the task function will be an instance of the Pydantic model defined for the pipeline's input parameters.

```python
@task
async def my_task(params: InputParams):
  result = params.some_value + 8
```

--------------------------------

### Register Pipeline with Daily Schedule

Source: https://lucafaggianelli.com/plombery/triggers

Demonstrates how to register a Plombery pipeline with a daily schedule using `apscheduler.triggers.interval.IntervalTrigger`. This involves defining a `Trigger` object with a specific schedule and passing it to the `register_pipeline` function.

```python
from apscheduler.triggers.interval import IntervalTrigger
from plombery import register_pipeline, Trigger

register_pipeline(
    id="sales_pipeline",
    tasks=[get_sales_data],
    triggers=[
        Trigger(
            id="daily",
            description="Run the pipeline every day",
            schedule=IntervalTrigger(
                days=1,
            ),
        ),
    ],
)
```

--------------------------------

### Passing Parameters via HTTP Trigger

Source: https://lucafaggianelli.com/plombery/triggers

When triggering a pipeline using the HTTP trigger, input parameters can be provided by including them as a JSON body in the POST request. This allows for programmatic control over pipeline inputs.

```APIDOC
HTTP Trigger with Parameters:
  Method: POST
  URL: <pipeline_url>
  Description: Triggers the pipeline execution with specified input parameters.
  Request Body:
    - Content-Type: application/json
    - Body:
      {
        "parameter_name_1": "value1",
        "parameter_name_2": "value2"
      }
  Parameter Validation:
    - Parameters must conform to the Pydantic BaseModel defined for the pipeline.
```

--------------------------------

### Register Pipeline for SSL Certificate Expiration Check

Source: https://lucafaggianelli.com/plombery/recipes/certificates-expiration

This snippet demonstrates how to define and register a Plombery pipeline to check SSL certificate expiration for a list of hostnames. It includes setting up tasks and triggers for each hostname, with a weekly schedule.

```python
hostnames = [
    "google.com",
    "velvetlab.tech",
]

register_pipeline(
    id="check_ssl_certificate",
    name="Check SSL certificate",
    description="""Check if the SSL certificate of a website has expired""",
    tasks=[check_certificate_expiration],
    triggers=[
        # Create 1 trigger per each host to check
        Trigger(
            id=f"check-{host}",
            name=host,
            description="Run the pipeline every week",
            params=InputParams(hostname=host),
            schedule=IntervalTrigger(
                weeks=1,
            ),
        )
        for host in hostnames
    ],
    params=InputParams,
)
```

--------------------------------

### Plombery Task Logging with get_logger

Source: https://lucafaggianelli.com/plombery/tasks

Shows how to use the `get_logger` function within a Plombery task to collect and display pipeline logs. The logger is obtained by calling `get_logger()` inside the task function.

```python
from plombery import get_logger

@task
def my_task():
  logger = get_logger()
  logger.debug("Hey greetings!")
```

--------------------------------

### Invalid Trigger Parameter Configuration

Source: https://lucafaggianelli.com/plombery/triggers

Illustrates an invalid way to configure trigger parameters where a mandatory parameter (`past_days`) is missing, leading to an error. This highlights the importance of providing all required parameters defined in the input model.

```python
Trigger(
    # ...
    params={
        "convert_currency": True,
    }
)
```

--------------------------------

### Check SSL Certificate Expiration Task

Source: https://lucafaggianelli.com/plombery/recipes/certificates-expiration

This Python code defines an asynchronous task for Plombery that checks the expiration date of an SSL certificate for a given hostname. It retrieves certificate information, compares the expiration date with the current time, and raises an exception if the certificate has expired or is nearing expiration.

```python
@task
async def check_certificate_expiration(params: InputParams):
    logger = get_logger()
    now = datetime.utcnow()

    info = get_certificate_info(params.hostname)
    expiration: datetime = info.get("notAfter")

    if expiration <= now:
        raise Exception(f"The certificate expired on {expiration}")

    expires_in = expiration - now

    if expires_in.days < EXPIRATION_WARNING_THRESHOLD:
```

--------------------------------

### Incorrect Usage of get_logger Outside Tasks

Source: https://lucafaggianelli.com/plombery/tasks

Illustrates the incorrect way to use `get_logger`. The `get_logger` function is specifically designed to work only within Plombery task functions and will not function correctly if called outside of them.

```python
# ❌ Don't do this
logger = get_logger()
def my_task():
  logger.debug("Hey greetings!")
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.