### Install airless-core using uv

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/quickstart.md

Installs the `airless-core` package using the `uv` package manager. It initializes a bare environment without a workspace and then adds the specified package.

```bash
uv init --bare --no-workspace
uv add airless-core
```

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

### Install airless-core using pip and venv

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/quickstart.md

Installs the `airless-core` package using Python's built-in `venv` module and `pip`. This involves creating a virtual environment, activating it, and then installing the package.

```bash
python -m venv .venv
# Activate venv in Mac / Linux
source .venv/bin/activate
# Windows CMD: .venv\Scripts\activate.bat
# Windows PowerShell: .venv\Scripts\Activate.ps1
pip install airless-core
```

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

### Create Hook and Operator Directories

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/quickstart.md

Creates the necessary directory structure for the project, specifically the `hook` and `operator` folders, which are essential for organizing the Airless application components.

```bash
mkdir hook operator
```

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

### Install Airless Packages (uv)

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

Installs airless-core, GCP dependencies, functions-framework, requests, and google-cloud-pubsub using uv. It also generates a requirements.txt file for Cloud Function deployment.

```bash
# Create a bare virtual environment (no pyproject.toml needed for this simple example)
uv venv .venv
source .venv/bin/activate # Or .\venv\Scripts\activate.bat / Activate.ps1 on Windows
# Install necessary packages
uv pip install airless-core airless-google-cloud-core airless-google-cloud-pubsub functions-framework requests google-cloud-pubsub
# Generate requirements.txt for Cloud Function deployment
uv pip freeze > requirements.txt
```

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

### Environment Variables for Airless

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/quickstart.md

Defines environment variables for the Airless project, including application environment and logging level. These are typically loaded from a .env file.

```env
ENV=dev
LOG_LEVEL=DEBUG
```

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

### Makefile for Running Airless Operator

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/quickstart.md

Provides a Makefile to execute the Airless WeatherOperator. The `run` target uses `python -c` to instantiate and execute the operator with specified parameters. Supports optional environment variable export.

```bash
run:
    @python -c "from operator.weather import WeatherOperator; WeatherOperator().execute(request={'data': {'request_type': 'temperature', 'lat': 40.7128, 'lon': -74.0060}})"
```

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

### Install Airless Packages (pip + venv)

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

Installs airless-core, GCP dependencies, functions-framework, requests, and google-cloud-pubsub using pip and venv. It also generates a requirements.txt file for Cloud Function deployment.

```bash
python -m venv .venv
# Activate venv in Mac / Linux
source .venv/bin/activate
# Windows CMD: .venv\Scripts\activate.bat
# Windows PowerShell: .venv\Scripts\Activate.ps1
pip install airless-core airless-google-cloud-core airless-google-cloud-pubsub functions-framework requests google-cloud-pubsub
pip freeze > requirements.txt
```

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

### Python API Hook for Weather Data

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/quickstart.md

Implements a hook to fetch current weather data using the Open-Meteo API. It inherits from BaseHook and uses the requests library to make API calls. Input is latitude and longitude, output is temperature in Celsius. Handles potential request errors by raising exceptions.

```python
import requests

from airless.core.hook import BaseHook


class ApiHook(BaseHook):
    """A simple hook to simulate fetching weather data."""

    def __init__(self):
        """Initializes the WeatherApiHook."""
        super().__init__()
        self.base_url = 'https://api.open-meteo.com/v1/forecast'

    def get_temperature(self, lat: float, lon: float) -> float:
        """
        Fetch the current temperature for a given city.

        Args:
            lat (float): The latitude of the city.
            lon (float): The longitude of the city.

        Returns:
            float: The current temperature in Celsius.

        Raises:
            ValueError: If the latitude or longitude is empty or invalid.
        """
        params = {
            'latitude': lat,
            'longitude': lon,
            'current': 'temperature_2m' 
        }
        with requests.Session() as session:
            response = session.get(
                self.base_url,
                params=params
            )
            response.raise_for_status()
            data = response.json()
            self.logger.debug(f"Response: {data}")

            temperature = data['current']['temperature_2m']

            return temperature
```

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

### Bash: Initialize Terraform

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

Initializes the Terraform working directory, downloading necessary providers and modules. This is a prerequisite for other Terraform commands like `plan` and `apply`.

```bash
terraform init
```

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

### Create Project Structure

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

Creates the necessary directories and empty files for the Airless project, including hook, operator, main Python files, Terraform configurations, Makefile, and .env file.

```bash
mkdir hook operator
touch hook/api.py operator/weather.py main.py main.tf function.tf variables.tf Makefile .env
# requirements.txt should already exist from the previous step
```

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

### Bash: Plan Terraform Deployment

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

Generates an execution plan for Terraform, showing which resources will be created, modified, or destroyed. Recommended for review before applying changes.

```bash
terraform plan
```

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

### Create Project Directories and Files (Bash)

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/multistep.md

Shell commands to create the necessary directories and empty files for the Airflow GCP workflow project structure. This sets up the foundational files for the hook, operator, Terraform configuration, main application, and requirements.

```bash
mkdir hook operator terraform
touch hook/api.py operator/weather.py terraform/main.tf terraform/function.tf terraform/variables.tf main.py requirements.txt Makefile .env
```

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

### Bash: Apply Terraform Deployment

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

Applies the Terraform execution plan, creating or updating infrastructure resources on Google Cloud Platform. This command packages code, uploads it, and provisions the defined resources.

```bash
terraform apply
```

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

### ApiHook for Weather Data Fetching (Python)

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

Defines an ApiHook class that inherits from BaseHook to fetch weather data from the Open-Meteo API. It includes methods for initializing the API URL, making GET requests, and handling responses and errors. It uses the requests library and provides a logger.

```python
import requests

from airless.core.hook import BaseHook


class ApiHook(BaseHook):
    """A simple hook to simulate fetching weather data."""

    def __init__(self):
        """Initializes the WeatherApiHook."""
        super().__init__()
        self.base_url = 'https://api.open-meteo.com/v1/forecast'

    def get_temperature(self, lat: float, lon: float) -> float:
        """
        Fetch the current temperature for a given city.

        Args:
            lat (float): The latitude of the city.
            lon (float): The longitude of the city.

        Returns:
            float: The current temperature in Celsius.

        Raises:
            requests.exceptions.RequestException: If the API request fails.
        """
        params = {
            'latitude': lat,
            'longitude': lon,
            'current': 'temperature_2m'
        }
        with requests.Session() as session:
            response = session.get(
                self.base_url,
                params=params,
                timeout=10 # Add a timeout
            )
            response.raise_for_status() # (3)!
            data = response.json()
            self.logger.debug(f"Response: {data}") # (4)!

            temperature = data['current']['temperature_2m']

            return temperature

```

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

### Google Cloud Function Entry Point (Python)

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/multistep.md

This Python script serves as the entry point for a Google Cloud Function. It utilizes the functions-framework library and dynamically imports the specific operator class based on the OPERATOR_IMPORT environment variable, allowing for flexible function logic.

```python
import functions_framework
import os

from airless.core.utils import get_config

```

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

### Python Requirements for Cloud Function

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/multistep.md

Specifies the Python packages required for the Cloud Function to run. The version constraint ensures compatibility. This file is typically read by a package manager like pip.

```text
airless-google-cloud-core~=0.1.2
```

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

### Install airless-captcha and Airless

Source: https://github.com/astercapital/airless/blob/main/packages/airless-captcha/README.md

This code snippet shows the necessary pip commands to install the airless-captcha package along with its core dependency, Airless. Ensure you are using a compatible version of Airless (>=0.0.69).

```shell
pip install airless>=0.0.69
pip install airless-captcha
```

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

### Python Project Dependencies (Text)

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

The requirements.txt file lists the Python dependencies necessary for the Cloud Function. It should be generated using pip freeze and includes the 'airless-google-cloud-core' package along with any other direct or transitive dependencies.

```text
airless-google-cloud-core==<version>
# Add any other direct or transitive dependencies listed by freeze

```

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

### Project Structure for Airflow GCP Deployment

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/multistep.md

Defines the directory and file structure required for the Airflow workflow deployment on GCP. This includes separate directories for hooks, operators, and Terraform configurations, along with main Python files and configuration files.

```text
.
├── hook
│   └── api.py           # Hook to interact with external APIs
├── operator
│   └── weather.py       # Operator containing business logic
├── terraform
│   ├── main.tf          # Terraform main configuration (provider, backend, archive)
│   ├── function.tf      # Terraform resources for Pub/Sub and Cloud Function
│   └── variables.tf     # Terraform input variables
├── main.py              # Cloud Function entry point
├── requirements.txt     # Python dependencies
├── Makefile             # Helper commands for deployment and triggering
└── .env                 # Environment variables for local Terraform execution
```

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

### Run Operator Locally with Makefile

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

This Makefile target `run` executes a Python command to instantiate and execute the WeatherOperator locally. It requires Python and the operator library to be available.

```makefile
run:
  @python -c "from operator.weather import WeatherOperator; WeatherOperator().execute({'request_type': 'temperature', 'lat': 51.5074, 'lon': -0.1278})"
```

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

### Local Environment Variables Configuration

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/multistep.md

Defines environment variables used primarily for local testing and Terraform variable injection. Sensitive information should not be committed. Variables cover environment naming, logging levels, and GCP project details.

```dotenv
# Environment Name (used for naming resources)
ENV=dev

# Logging Level for local testing (Cloud Function level set in Terraform)
LOG_LEVEL=DEBUG

# --- GCP Configuration ---
# Replace with your actual GCP Project ID
GCP_PROJECT="your-gcp-project-id"
QUEUE_TOPIC_ERROR="dev-airless-error"
```

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

### Gcloud CLI: Publish to Pub/Sub Topic

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

Manually publishes a message to a specified Pub/Sub topic using the gcloud command-line interface. This is useful for testing the function trigger without waiting for the scheduler.

```bash
gcloud pubsub topics publish dev-weather-api --message '{"request_type": "temperature", "lat": 51.5074, "lon": -0.1278}'
```

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

### Terraform Main Configuration for Source Code Archiving

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/multistep.md

This Terraform configuration defines the process for archiving the Cloud Function's source code into a zip file. It uses the `archive_file` data source to create a zip archive of the current directory, applying specified exclusions.

```terraform
# Archive the source code directory into a zip file
data "archive_file" "source" {
  type        = "zip"
  source_dir  = "${path.module}/" # Zips the current directory
  output_path = "/tmp/${var.env}-${var.function_name}-source.zip"
  excludes    = var.source_archive_exclude

  # Include necessary files/dirs explicitly if source_dir is broader
  # For this structure, source_dir = "." works fine with excludes.
}
```

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

### Upload Zipped Source Code to GCS (Terraform)

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/multistep.md

Defines a Google Cloud Storage bucket object to store the zipped source code for the Cloud Function. It uses a generated name based on environment, function name, and MD5 hash for uniqueness. Depends on the `data.archive_file` and `google_storage_bucket` resources.

```Terraform
resource "google_storage_bucket_object" "zip" {
  name   = "${var.env}-${var.function_name}-src-${data.archive_file.source.output_md5}.zip"
  bucket = google_storage_bucket.function_source_bucket.name # Use the created bucket name
  source = data.archive_file.source.output_path
}
```

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

### Execute Chained Messages with Airless Operator

Source: https://context7.com/astercapital/airless/llms.txt

Demonstrates how to execute chained messages using the Airless operator. This example defines a data structure for a temperature request, including metadata for a subsequent 'store-temperature' operation. The operator then processes this chained data.

```python
chained_data = {
    'request_type': 'temperature',
    'lat': 51.5074,
    'lon': -0.1278,
    'metadata': {
        'run_next': [
            {
                'topic': 'store-temperature',
                'data': {
                    'request_type': 'store',
                    'location': 'London'
                }
            }
        ]
    }
}

operator.execute(data=chained_data, topic='test-topic')
```

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

### Dynamically Import Operator in Python Cloud Function

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/multistep.md

This snippet demonstrates how to dynamically import an operator class based on an environment variable at runtime. It's designed for a Cloud Function entry point that routes events to the appropriate operator. Dependencies include the 'functions-framework' library.

```python
import functions_framework

# Assume get_config is defined elsewhere to fetch environment variables
# Assume OperatorClass will be dynamically imported and available

# Dynamically import the operator based on environment variable
exec(f'{get_config("OPERATOR_IMPORT")} as OperatorClass') # (1)!

@functions_framework.cloud_event # (2)!
def route(cloud_event):
    """
    Cloud Function entry point triggered by a Pub/Sub event.
    Dynamically routes the event to the appropriate Airless operator.
    """
    # Instantiate the dynamically loaded operator class
    operator_instance = OperatorClass()
    # Run the operator with the incoming event data
    operator_instance.run(cloud_event) # (3)!
```

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

### Python Operator for Weather Data Retrieval

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/quickstart.md

Defines an operator that utilizes the ApiHook to retrieve weather information. It inherits from BaseOperator and routes requests based on 'request_type'. Input is a dictionary containing request details, output is None. Uses the built-in logger for debugging.

```python
from airless.core.operator import BaseOperator

from hook.api import ApiHook


class WeatherOperator(BaseOperator):
    """A simple operator to simulate fetching weather data."""

    def __init__(self):
        """Initializes the WeatherOperator."""
        super().__init__()
        self.api_hook = ApiHook()

    def execute(self, data: dict, topic: str) -> None:
        """Define which method to call based on the request type."""
        request_type = data['request_type']

        if request_type == 'temperature':
            self.get_temperature(data, topic)
        else:
            raise Exception(f'Request type {request_type} not implemented')

    def get_temperature(self, data: dict, topic: str) -> None:
        """Fetch the current temperature for a given city."""
        lat = data['lat']
        lon = data['lon']
        temperature = self.api_hook.get_temperature(lat, lon)

        self.logger.debug(f"Temperature: {temperature}")
```

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

### Terraform Provider Configuration

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/core-infrastructure.md

Defines the cloud provider and region for the Terraform deployment. This example specifies the Google Cloud provider, using variables for project ID and region.

```terraform
provider "google" {
  project = var.project_id
  region  = var.region
}
```

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

### Bash: Destroy Terraform Resources

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

Removes all GCP resources created by the Terraform configuration. This command prompts for confirmation before proceeding with the deletion.

```bash
terraform destroy
```

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

### GCP Cloud Function Entry Point (Python)

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/simple.md

The main.py file serves as the entry point for the GCP Cloud Function. It utilizes the functions_framework to handle incoming Cloud Events, dynamically imports the specified operator class based on an environment variable, instantiates it, and executes the operator's run method. It requires the 'functions-framework' library.

```python
import functions_framework
import os

from airless.core.utils import get_config

exec(f'{get_config("OPERATOR_IMPORT")} as OperatorClass')

@functions_framework.cloud_event
def route(cloud_event):
    """
    Cloud Function entry point triggered by a Pub/Sub event.
    Dynamically routes the event to the appropriate Airless operator.
    """
    operator_instance = OperatorClass()
    operator_instance.run(cloud_event)

```

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

### Terraform Root Main Configuration

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/core-infrastructure.md

Sets up core infrastructure and calls the Airless module. It defines a Google Cloud Storage bucket for function code and a Pub/Sub topic, then instantiates the 'airless_core' module, passing necessary variables and resource references.

```terraform
resource "google_storage_bucket" "function_code_bucket" {
  name     = "${var.env}-airless-function-code"
  location = var.region
}

resource "google_pubsub_topic" "pubsub_to_bq" {
  name = "${var.env}-pubsub-to-bq"
}

module "airless_core" {
  source = "./modules/airless-core"

  project_id = var.project_id
  region     = var.region
  env        = var.env
  log_level  = var.log_level

  function_bucket = {
    id   = google_storage_bucket.function_code_bucket.id
    name = google_storage_bucket.function_code_bucket.name
  }

  queue_topic_pubsub_to_bq = {
    id   = google_pubsub_topic.pubsub_to_bq.id
    name = google_pubsub_topic.pubsub_to_bq.name
  }

  source_archive_exclude = [
    ".*",
    "__pycache__",
    "*.pyc"
  ]
}
```

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

### Define Pub/Sub Topic and Cloud Function (Terraform)

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/multistep.md

Configures a Google Cloud Pub/Sub topic for triggering the function and the Cloud Function resource itself. It specifies the runtime (Python 3.12), entry point, source code location from GCS, service configuration (memory, timeout, environment variables), and the Pub/Sub event trigger with retry policy.

```Terraform
# Pub/Sub Topic to trigger the function
resource "google_pubsub_topic" "main_topic" {
  name = "${var.env}-${var.function_name}"
}

# The Cloud Function resource
resource "google_cloudfunctions2_function" "main_function" {
  name     = "${var.env}-${var.function_name}"
  location = var.region
  description = "Airless function to fetch weather data from API"

  build_config {
    runtime     = "python312" # Or python310, python311, python312
    entry_point = "route"    # Matches the function name in main.py
    source {
      storage_source {
        bucket = google_storage_bucket_object.zip.bucket # Get bucket from the uploaded object
        object = google_storage_bucket_object.zip.name   # Get object name from the uploaded object
      }
    }
  }

  service_config {
    max_instance_count = 3 # Limit concurrency
    available_memory   = "256Mi"
    timeout_seconds    = 60
    # Define environment variables needed by the function and airless core/gcp libs
    environment_variables = {
      ENV                  = var.env
      LOG_LEVEL            = var.log_level
      GCP_PROJECT          = var.project_id # Airless GCP libs might need this
      GCP_REGION           = var.region     # Airless GCP libs might need this
      OPERATOR_IMPORT      = "from operator.weather import WeatherOperator"
      QUEUE_TOPIC_ERROR    = var.pubsub_topic_error_name # For base operator error routing
      # Add any other specific env vars your operator/hook might need
    }
    # ingress_settings               = "ALLOW_ALL" # Default - Allow public access if needed (not for PubSub trigger)
    # all_traffic_on_latest_revision = true
  }

  # Configure the trigger (Pub/Sub topic)
  event_trigger {
    trigger_region = var.region # Can differ from function region if needed
    event_type     = "google.cloud.pubsub.topic.v1.messagePublished"
    pubsub_topic   = google_pubsub_topic.main_topic.id
    retry_policy   = "RETRY_POLICY_RETRY" # Retry on failure
  }
}
```

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

### Python Cloud Function Dependencies

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/core-infrastructure.md

This file lists the Python packages required for the Cloud Functions. These dependencies are installed by Google Cloud when building the function's runtime environment. It includes specific versions for various Airless Google Cloud and communication libraries.

```text
airless-google-cloud-secret-manager~=0.1.0
airless-google-cloud-storage~=0.1.0
airless-google-cloud-bigquery~=0.1.0
airless-slack~=0.1.0
airless-email~=0.1.0
```

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

### Cloud Function Entry Point with Dynamic Operator Import (Python)

Source: https://github.com/astercapital/airless/blob/main/docs/tutorials/gcp/core-infrastructure.md

This Python code serves as the entry point for Cloud Functions within the Airless module. It dynamically imports an operator class based on an environment variable (`OPERATOR_IMPORT`) and then uses this operator to process incoming Cloud Events. This allows for flexible routing of events to different operator implementations.

```python
import functions_framework
import os

from airless.core.utils import get_config

# Dynamically import the operator based on environment variable
exec(f'{get_config("OPERATOR_IMPORT")} as OperatorClass')

@functions_framework.cloud_event
def route(cloud_event):
    """
    Cloud Function entry point triggered by a Pub/Sub event.
    Dynamically routes the event to the appropriate Airless operator.
    """
    # Instantiate the dynamically loaded operator class
    operator_instance = OperatorClass()
    # Run the operator with the incoming event data
    operator_instance.run(cloud_event)
```