Transformers¶

Transformers modify data between extraction and loading.

Overview¶

from pycharter import Rename, Filter, AddField, Drop, Select, Convert, CustomFunction

Built-in Transformers¶

Rename¶

Rename fields in records:

from pycharter import Rename

# Single field
transform = Rename({"old_name": "new_name"})

# Multiple fields
transform = Rename(
    {"userName": "user_name", "userEmail": "email", "createdAt": "created_at"}
)

Filter¶

Filter records based on conditions:

from pycharter import Filter

# Lambda function
transform = Filter(lambda r: r.get("active", False))

# Multiple conditions
transform = Filter(lambda r: r["age"] >= 18 and r["status"] == "active")

# Using expression string
transform = Filter("active == true and age >= 18")

AddField¶

Add new fields to records:

from pycharter import AddField

# Static value
transform = AddField("source", "api")

# Current timestamp
transform = AddField("processed_at", "now()")

# Expression
transform = AddField("full_name", "concat(first_name, ' ', last_name)")
transform = AddField("is_adult", "age >= 18")

# Multiple fields
transform = AddField({"source": "api", "processed_at": "now()"})

Drop¶

Remove fields from records:

from pycharter import Drop

# Single field
transform = Drop(["password"])

# Multiple fields
transform = Drop(["password", "ssn", "internal_id", "temp_field"])

Select¶

Keep only specified fields:

from pycharter import Select

# Whitelist fields
transform = Select(["id", "name", "email", "created_at"])

Convert¶

Convert field types:

from pycharter import Convert

transform = Convert(
    {
        "age": "int",
        "price": "float",
        "active": "bool",
        "created_at": "datetime",
        "tags": "list",
    }
)

Default¶

Set default values for missing fields:

from pycharter import Default

transform = Default({"status": "pending", "priority": 0, "tags": []})

Map¶

Apply a function to each record:

from pycharter import Map


def process_record(record):
    record["name"] = record["name"].title()
    return record


transform = Map(process_record)

FlatMap¶

Map and flatten results:

from pycharter import FlatMap


def explode_tags(record):
    # Return multiple records from one
    for tag in record.get("tags", []):
        yield {**record, "tag": tag}


transform = FlatMap(explode_tags)

CustomFunction¶

Apply custom transformation logic:

from pycharter import CustomFunction


def enrich_user(records):
    for record in records:
        record["display_name"] = f"{record['first_name']} {record['last_name']}"
        record["email_domain"] = record["email"].split("@")[1]
    return records


transform = CustomFunction(enrich_user)

Chaining Transformers¶

Use the | operator to chain transformers:

from pycharter import Pipeline, Rename, Filter, AddField, Drop

pipeline = (
    Pipeline(extractor)
    | Rename({"userName": "user_name"})
    | Filter(lambda r: r.get("active"))
    | AddField("processed_at", "now()")
    | Drop(["password", "internal_id"])
    | loader
)

TransformerChain¶

Create reusable transformer chains:

from pycharter import TransformerChain, Rename, Filter, AddField

# Create reusable chain
user_transforms = TransformerChain(
    [
        Rename({"userName": "user_name"}),
        Filter(lambda r: r.get("active")),
        AddField("processed_at", "now()"),
    ]
)

# Use in pipeline
pipeline = Pipeline(extractor) | user_transforms | loader

Config-Driven Transformers¶

Define transformers in YAML:

pipeline.yaml (transform step operations)

rename:
  userName: user_name
  userEmail: email

filter:
  expression: "active == true"

add:
  processed_at: "now()"
  source: "api"

drop:
  - password
  - internal_id

convert:
  age: int
  price: float

Expression Syntax¶

Supported expressions for AddField and Filter:

Expression	Description	Example
`now()`	Current timestamp	`"processed_at": "now()"`
`concat(...)`	String concatenation	`"full_name": "concat(first, ' ', last)"`
`field >= value`	Comparison	`"is_adult": "age >= 18"`
`field == value`	Equality	`"is_active": "status == 'active'"`
`field and field`	Logical AND	`"active and verified"`
`field or field`	Logical OR	`"admin or moderator"`
`not field`	Logical NOT	`"not deleted"`

Custom Transformers¶

Create custom transformers by extending BaseTransformer:

from pycharter.pipeline_generator.transformers import BaseTransformer
from typing import List, Dict, Any


class UppercaseNames(BaseTransformer):
    def __init__(self, fields: List[str]):
        self.fields = fields

    def transform(self, records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        for record in records:
            for field in self.fields:
                if field in record and isinstance(record[field], str):
                    record[field] = record[field].upper()
        return records


# Use in pipeline
pipeline = Pipeline(extractor) | UppercaseNames(["name", "city"]) | loader