Adding Function Bricks

In the previous section, we discussed the general structure of a CODED FLOWS package and the requirement to place all Bricks in its corresponding folder. In this section, we will detail how to add and configure those Bricks, starting with a tutorial to guide you through creating your first Brick, followed by a reference section with detailed configuration options.

Building Your First Brick

Objective

We will create a CODED FLOWS Brick that reads an image from a text-based path and returns it in one of the following selected formats: NumPy array, Pillow image, bytes, or BytesIO.

This Brick will provide the following visual node in CODED FLOWS:

Installing Dependencies

We need to install the official CODED FLOWS package coded-flows to define the types for the inputs and outputs of a Brick.

# Using pip
pip install coded-flows

# Using uv
uv pip install coded-flows

# Using conda
conda install -c conda-forge coded-flows

Creating The Brick File

Our Brick function will be named read_image, so we need to create a Python script also named read_image.py and place it inside the Bricks folder of the package.

The Brick Main Function

We will start with a core Python function and complete each step to transform it into its final, fully adapted form for CODED FLOWS.

from PIL import Image

def read_image(path):
    image = Image.open(path)
    return image

This function alone is sufficient to define a Brick. CODED FLOWS will automatically detect the presence of one input, path, and one output, image, assigning both the type Any. This means there will be no validation for connection constraints on the related node.

We can enhance our Brick by adding metadata through declaring the coded_flows_metadata variable as follows:

from PIL import Image

coded_flows_metadata = {
    "display_name": "Read Image",
    "description": "Reads an image from a file path and returns it in a specified format",
    "icon": "photo-filled",
    "requirements": ["numpy", "pillow"]
}

def read_image(path):
    image = Image.open(path)
    return image

• display_name: Specifies the name to be displayed. Otherwise, the default behavior is to display the Brick's main function name without underscores (in this case, read image).
• description: A brief description of the Brick's functionality, displayed in both the package repository and the CODED FLOWS app UI.
• icon: The icon associated with the Brick. We use the icon values of tabler.io.
• requirements: A list of Python libraries required for the Brick to function.

Our Brick is still missing a key element mentioned in the objective: the ability to interactively select the type of output we want. We can add this interactivity by using options, which we will define in the metadata and pass as an expected input named options. The options input name is reserved and is only used if defined in the coded_flows_metadata. We can then access the options menu by click at its corresponding button.

We have now made three updates to our Brick script:

• Adding the options configuration to coded_flows_metadata, defining one option element of type select with four available choices, where the array choice is the default value.
• Adding the corresponding input options to the read_image function, a reserved name expected by the CODED FLOWS app.
• Updating the read_image function to account for the output_type value in options and return the requested format accordingly.

import io
import numpy as np
from PIL import Image


coded_flows_metadata = {
    "display_name": "Read Image",
    "description": "Reads an image from a file path and returns it in a specified format",
    "icon": "photo-filled",
    "requirements": ["numpy", "pillow"],
    "options": [
        {
            "name": "output_type",
            "display_name": "Output Type",
            "type": "select",
            "choices": ["array", "pil", "bytes", "bytesio"],
            "default": "array",
        }
    ],
}


def read_image(path, options):
    output_type = options["output_type"]

    image = Image.open(path)

    if output_type == "array":
        image = np.array(image)
    elif output_type == "bytes":
        with io.BytesIO() as buffer:
            image.save(buffer, format=image.format)
            image = buffer.getvalue()
    elif output_type == "bytesio":
        buffer = io.BytesIO()
        image.save(buffer, format=image.format)
        buffer.seek(0)
        image = buffer

    return image

Currently, our Brick lacks type constraints on its input and output, as they default to the Any type. This allows them to connect to any input or output of other Bricks in the flow, which can lead to errors. To address this, we will use the official coded-flows package to import the necessary type hints and apply them to the corresponding elements: the path input, expected to be a text-based path, will use the Str type, and the output, which can be one of four types (PILImage, NDArray, Bytes, BytesIOType), will be combined using Union (alternatively, the | operator can be used). The options input does not require type hinting, as it is a reserved input name managed by CODED FLOWS.

You can find the complete list of available types at the Available Type Hints part of this page.

The complete form of our Brick script is as follows:

import io
import numpy as np
from PIL import Image
from typing import Union
from coded_flows.types import Str, PILImage, NDArray, Bytes, BytesIOType


coded_flows_metadata = {
    "display_name": "Read Image",
    "description": "Reads an image from a file path and returns it in a specified format",
    "icon": "photo-filled",
    "requirements": ["numpy", "pillow"],
    "options": [
        {
            "name": "output_type",
            "display_name": "Output Type",
            "type": "select",
            "choices": ["array", "pil", "bytes", "bytesio"],
            "default": "array",
        }
    ],
}


def read_image(path: Str, options) -> Union[PILImage, NDArray, Bytes, BytesIOType]:
    output_type = options["output_type"]

    image = Image.open(path)

    if output_type == "array":
        image = np.array(image)
    elif output_type == "bytes":
        with io.BytesIO() as buffer:
            image.save(buffer, format=image.format)
            image = buffer.getvalue()
    elif output_type == "bytesio":
        buffer = io.BytesIO()
        image.save(buffer, format=image.format)
        buffer.seek(0)
        image = buffer

    return image

Brick Configuration

Configure the structure, interface, and behavior of a Brick, including its metadata, inputs, outputs, types, and logging.

Brick Metadata

You can add metadata to a Brick to enrich the generated visual node by declaring the coded_flows_metadata dictionary as follows:

# imports
...

coded_flows_metadata = {
    "type": "function",
    "display_name": "Brick Display Name",
    "description": "Short description of the Brick.",
    "icon": "package",
    "requirements": ['numpy', 'pandas']
}

# the rest of the script
...

coded_flows_metadata takes the following parameters:

Interactive Options

As demonstrated in the example presented in this part of the guide, we can introduce interactivity to our Bricks by using the options parameter in coded_flows_metadata. Interactivity refers to the ability to select values, input text, or define other dynamic parameters for the Brick.

To make the defined options effective, you need to include the reserved input name (otpions) in the main function's inputs as follows:

coded_flows_metadata = {
    ...
    "options": [...]
    ...
}

def brick_name(input1, input2, options):
    ...

To access an option's value, retrieve the corresponding key representing its name from the options dictionary as follows:

coded_flows_metadata = {
    ...
    "options": [...]
    ...
}

def brick_name(input1, input2, options):
    variable = options['option_name'] # option value extraction
    ...

The options parameter in coded_flows_metadata contains a list of dictionaries. There are seven available types:

• toggle: A boolean value represented as a toggle.
• select: A value selected from a predefined list.
• number: A float value.
• integer: An integer value.
• input: A short text input.
• textarea: A long text input.
• textual_values: A user-defined list of text values.

The seven types share three common parameters: name, which defines the Pythonic name used to extract the corresponding option value; display_name, which specifies the name to display in the visual node; and type, which indicates one of the seven available types:

coded_flows_metadata = {
    ...
    "options": [
        {
            "name": "pythonic_name",
            "display_name": "Name to Display",
            "type": "toggle", # toggle, select, number, integer, input, textarea or textual_values
            ...
        }
    ]
    ...
}

Below is an example of the corresponding dictionary for each type within the options list:

  {
      "name": "is_intensity",
      "display_name": "Display intensity",
      "type": "toggle",
      "default": False,
  }

  {
      "name": "output_type",
      "display_name": "Output Type",
      "type": "select",
      "choices": ["array", "pil", "bytes", "bytesio"],
      "default": "array",
  }

  {
      "name": "lr",
      "display_name": "Learning Rate",
      "type": "number",
      "step": 0.01, # default 1
      "max": 1, # default 100
      "min": 0.0001, # default -100
      "default": 0.1, # default 0
  }

  {
      "name": "nb_records",
      "display_name": "Number of records",
      "type": "integer",
      "step": 1, # default 1
      "max": 10000, # default 100
      "min": 5, # default -100
      "default": 10, # default 0
  }

  {
      "name": "separator",
      "display_name": "Text Separator",
      "type": "input",
      "default": "_",
      "max_characters": 3, # default 50
  }

  {
      "name": "reference_text",
      "display_name": "Reference Words",
      "type": "textarea",
      "default": "this is default text",
      "max_characters": 500, # default 1000
  }

  {
      "name": "columns",
      "display_name": "List of Columns",
      "type": "textual_values",
      "values": ["id", "date"], # default []
  }

The step parameter for number and integer types defines the increment or decrement value when clicking the corresponding visual + and - buttons.

Defining Inputs and Outputs

CODED FLOWS automatically extracts the inputs and outputs from your main Brick function, with only a few minor constraints to follow.

Inputs

All your inputs are translated into Brick entries, so including input arguments in your main function is sufficient to define the node's entries.

def brick_name(input1, input2, input3):
    ...

If your main function has no input arguments, your Brick will not expect any entries.

# a Brick node with no entries
def brick_name():
    ...

Outputs

There are three cases for Brick outputs: it has none, a single output, or multiple outputs.

• If there are no outputs, simply omit the return statement in your main function.

• For a single output, the output is extracted from the return statement. In this example, it will be named variable_name:

  def brick_name(...):
      ...
      return variable_name
  

  If the returned expression is not a variable, the output will be named output:

  def brick_name(...):
      ...
      return [1, 2, 3] # This output will be named `output`
  

• For multiple outputs, CODED FLOWS expects a tuple to be returned. In this example, the extracted outputs are named variable_name1, variable_name2, and variable_name3, respectively:

  def brick_name(...):
      ...
      return variable_name1, variable_name2, variable_name3
  

  If the returned expressions aren't variables, the outputs will be named output, output_1 and output_2:

  def brick_name(...):
      ...
      return 100, [1, 2, 3], "text"
  

Optional Inputs

To define an optional input in a Brick, we simply assign a default value to its corresponding function input, as shown below:

def create_file(text_content, file_path, file_name='file.txt'):
    ...

This way, if the input is not connected to any other Brick's output, no error will be raised.

You can visually distinguish an optional input by the brackets surrounding its name:

If you have a default value and you defined options for this Brick, you must assign to options a default value of None to prevent errors in your Python function:

def create_file(text_content, file_path, file_name = 'file.txt', options = None):
    ...

Defining Input and Output Types

Defining type hints for your inputs and outputs enables CODED FLOWS to validate connections between Bricks, reducing the occurrence of errors and ensuring that each Brick is connected to a compatible one.

To apply type hints, you first need to install the official CODED FLOWS package:

# Using pip
pip install coded-flows

# Using uv
uv pip install coded-flows

# Using conda
conda install -c conda-forge coded-flows

Then, you can import the type hints as follows:

from coded_flows.types import Str, PILImage, NDArray, Bytes, BytesIOType

You can use them in the same way as standard Python type hints:

from typing import Union, Tuple
from coded_flows.types import Str, PILImage, NDArray, Bytes, BytesIOType

def image_rgb_channels(
    input_image: Union[Str, PILImage, NDArray, Bytes, BytesIOType], options
) -> Tuple[PILImage, PILImage, PILImage]:
    ...
    return r, g, b

If an input or output can have multiple possible types, use Union or the | operator. For multiple returned outputs, use Tuple. Union and Tuple can be imported from the native Python typing module.

Another example:

from coded_flows.types import Str, PILImage, NDArray, Bytes, BytesIOType

def read_image(path: Str, options) -> Union[PILImage, NDArray, Bytes, BytesIOType]:
    ...
    return image

You can find the complete list of available types at the Available Type Hints part of this page.

Brick Logging

CODED FLOWS displays the logs of your flow during each execution, allowing users to monitor its progress.

You can add this capability to your Brick by using Python's standard logging module:

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def brick_name(...):
    ...
    logger.info("Your log message to be displayed to the user.")
    ...

Available Type Hints

Here is a list of the available type hints that you can import from coded_flows.type. For each type, we provide a brief example of the expected values and for type hints that can be used as helpers, we also include a usage snippet.

Any

Represents any type of data. It can connect to all other types.

Null

Represents a null value.

# Example of `Null` expected values
value = None

DataSeries

Represents a series of data, typically used in data analysis.

# Example of `DataSeries` expected values.

# Pandas
import pandas as pd

# Create a Pandas Series
pandas_series = pd.Series([1, 2, 3, 4, 5], name='numbers')


# Polars
import polars as pl

# Create a Polars Series
polars_series = pl.Series("numbers", [1, 2, 3, 4, 5])

DataFrame

Represents a tabular data structure, commonly used in data science.

# Examples of `DataFrame` expected values.

# Pandas
import pandas as pd

# Create Pandas DataFrame
pandas_df = pd.DataFrame({
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [25, 30, 35],
    'city': ['New York', 'London', 'Paris']
})

# Polars
import polars as pl

# Create Polars DataFrame
polars_df = pl.DataFrame({
    'name': ['Alice', 'Bob', 'Charlie'],
    'age': [25, 30, 35],
    'city': ['New York', 'London', 'Paris']
})

ArrowTable

Represents a table structure in Apache Arrow format.

NDArray

Represents a multidimensional Numpy array, often used in numerical computations.

DataDict

Represents a dictionary-like structure where each string key maps to a list of values.

# Example of `DataDict` expected values
data = {
    "city": ["Tokyo", "Paris", "New York"],
    "country": ["Japan", "France", "United States"]
}

DataRecords

Represents a list of data records, typically key-value pairs.

# Example of `DataRecords` expected values
data = [
    {'city': 'Tokyo', 'country': 'Japan'},
    {'city': 'Paris', 'country': 'France'},
    {'city': 'New York', 'country': 'United States'}
]

Str

Represents a string value.

Compatible types: Str, Base64Str, CountryAlpha2, CountryAlpha3, CountryNumericCode, CountryShortName, EmailStr, Currency, Json, MacAddress

AnyStr

Represents either a string or bytes value.

Compatible types: AnyStr, Str, Base64Str, CountryAlpha2, CountryAlpha3, CountryNumericCode, CountryShortName, EmailStr, Bytes, Currency, Base64Bytes, Json, MacAddress

Base64Str

Represents a base64-encoded string.

CountryAlpha2

Represents a two-letter country code (ISO 3166-1 alpha-2).

This type is a Pydantic type and can be used as a helper.

from coded_flows.types import CountryAlpha2

country = CountryAlpha2("US")
country_alpha3 = country.alpha3
country_short_name = country.short_name
country_numeric_code = country.numeric_code

CountryAlpha3

Represents a three-letter country code (ISO 3166-1 alpha-3).

This type is a Pydantic type and can be used as a helper.

from coded_flows.types import CountryAlpha3

country = CountryAlpha3("USA")
country_alpha2 = country.alpha2
country_short_name = country.short_name
country_numeric_code = country.numeric_code

CountryNumericCode

Represents a numeric country code (ISO 3166-1 numeric).

This type is a Pydantic type and can be used as a helper.

from coded_flows.types import CountryNumericCode

country = CountryNumericCode(840)
country_alpha2 = country.alpha2
country_alpha3 = country.alpha3
country_short_name = country.short_name

CountryShortName

Represents a short name for a country.

This type is a Pydantic type and can be used as a helper.

from coded_flows.types import CountryShortName

country = CountryShortName("United States")
country_alpha2 = country.alpha2
country_alpha3 = country.alpha3
country_short_name = country.short_name

Currency

Represents a currency code (ISO 4217).

Bool

Represents a boolean value (True or False).

Datetime

Represents a date and time value.

# Example of `Datetime` expected values
from datetime import datetime

dt_value = datetime(2025, 5, 7, 14, 30, 15)

Date

Represents a date value.

Compatible types: Date, Datetime

# Example of `Date` expected values
from datetime import date

date_value = date(2025, 5, 7)

Time

Represents a time value.

# Example of `Time` expected values
from datetime import time

time_value = time(14, 30, 15)

Timedelta

Represents a duration or difference between two times.

# Example of `Timedelta` expected values
from datetime import timedelta

# timedelta: e.g., 3 days, 5 hours, 10 minutes
delta_value = timedelta(days=3, hours=5, minutes=10)

Int

Represents an integer value.

Compatible types: Int, PositiveInt, NegativeInt

Float

Represents a floating-point number.

Compatible types: Float, Longitude, Latitude, PositiveFloat, NegativeFloat, FiniteFloat, Int, PositiveInt, NegativeInt

Complex

Represents a complex number.

Compatible types: Complex, Int, Float, PositiveFloat, NegativeFloat, FiniteFloat, PositiveInt, NegativeInt

Decimal

Represents a decimal number with fixed precision.

Compatible types: Decimal, Int, PositiveInt, NegativeInt

Number

Represents a numeric value, which can be a boolean, integer, float, or complex number.

Compatible types: Bool, Int, Float, PositiveInt, NegativeInt, Longitude, Latitude, PositiveFloat, NegativeFloat, FiniteFloat, Complex

PositiveInt

Represents a positive integer value.

NegativeInt

Represents a negative integer value.

PositiveFloat

Represents a positive floating-point number.

NegativeFloat

Represents a negative floating-point number.

FiniteFloat

Represents a finite floating-point number (non-infinite, non-NaN).

List

Represents a list of values.

Compatible types: List, DataRecords

Tuple

Represents a tuple of values.

Deque

Represents a double-ended queue.

Set

Represents a set of unique values.

FrozenSet

Represents an immutable set of unique values.

Iterable

Represents an iterable collection of values.

Compatible types: Iterable, List, Tuple, Deque, Set, FrozenSet, DataSeries, DataFrame, NDArray, DataRecords, Bytes, Base64Bytes, AnyStr, Str, Base64Str, CountryAlpha2, CountryAlpha3, CountryNumericCode, CountryShortName, EmailStr, Currency, Json, MacAddress

Dict

Represents a dictionary of key-value pairs.

Compatible types: Dict, DataDict

Callable

Represents a callable object, such as a function.

IPvAnyAddress

Represents an IP address (either IPv4 or IPv6).

This type is a Pydantic type and can be used as a helper for validation and parsing of IP addresses.

from coded_flows.types import IPvAnyAddress

# Valid IP address (IPv4 or IPv6)
addr = IPvAnyAddress("192.168.0.1")

# You can also use IPv6 values
addr6 = IPvAnyAddress("2001:db8::1")

IPvAnyInterface

Represents an IP interface (address with subnet).

This type is a Pydantic type and can be used as a helper for validation and parsing of IP interfaces.

from coded_flows.types import IPvAnyInterface

# Valid IP address (IPv4 or IPv6)
iface = IPvAnyInterface("192.168.0.1/24")

# You can also use IPv6 values
iface6 = IPvAnyInterface("2001:db8::1/64")

IPvAnyNetwork

Represents an IP network.

This type is a Pydantic type and can be used as a helper for validation and parsing of IP networks.

from coded_flows.types import IPvAnyNetwork

# Valid IP address (IPv4 or IPv6)
net = IPvAnyNetwork("192.168.0.0/24")

# You can also use IPv6 values
net6 = IPvAnyNetwork("2001:db8::/64")

AnyUrl

Represents any URL string.

Compatible types: AnyUrl, AnyHttpUrl, HttpUrl, FileUrl, PostgresDsn, CockroachDsn, AmqpDsn, RedisDsn, MongoDsn, KafkaDsn, NatsDsn, MySQLDsn, MariaDBDsn

This type is a Pydantic type and can be used as a helper for validation and parsing of URLs.

from coded_flows.types import AnyUrl

# Direct validation of a URL string
try:
    url_value = AnyUrl("https://example.com")
    print(f"Valid URL: {url_value}")
except ValueError as e:
    print(f"Invalid URL: {e}")

# Access URL components
if url_value:
    print(f"Scheme: {url_value.scheme}")
    print(f"Host: {url_value.host}")
    print(f"Path: {url_value.path}")

AnyHttpUrl

Represents an HTTP or HTTPS URL.

Compatible types: AnyHttpUrl, HttpUrl

This type is a Pydantic type and can be used as a helper for validation and parsing of HTTP URLs.

from coded_flows.types import AnyHttpUrl

# Direct validation of an HTTP/HTTPS URL
try:
    http_url = AnyHttpUrl("https://api.example.com/data?query=test")
    print(f"Valid HTTP URL: {http_url}")
except ValueError as e:
    print(f"Invalid HTTP URL: {e}")

# Non-HTTP URLs will be rejected
try:
    AnyHttpUrl("ftp://example.com")
except ValueError as e:
    print(f"Rejected non-HTTP URL: {e}")

HttpUrl

Represents a strict HTTP URL.

This type is a Pydantic type and can be used as a helper for validation and parsing of HTTP URLs.

from coded_flows.types import HttpUrl

# Direct validation of a strict HTTP URL
try:
    url = HttpUrl("https://www.example.com/path")
    print(f"Valid HTTP URL: {url}")

    # Access parts of the URL
    print(f"Domain: {url.host}")
    print(f"Path: {url.path}")
except ValueError as e:
    print(f"Invalid HTTP URL: {e}")

FileUrl

Represents a file URL.

This type is a Pydantic type and can be used as a helper for validation and parsing of file URLs.

from coded_flows.types import FileUrl

# Direct validation of a file URL
try:
    file_url = FileUrl("file:///home/user/document.txt")
    print(f"Valid file URL: {file_url}")
    print(f"Path: {file_url.path}")
except ValueError as e:
    print(f"Invalid file URL: {e}")

# Non-file URLs will be rejected
try:
    FileUrl("https://example.com")
except ValueError as e:
    print(f"Rejected non-file URL: {e}")

PostgresDsn

Represents a PostgreSQL connection string (DSN).

This type is a Pydantic type and can be used as a helper for validation and parsing of PostgreSQL connection strings.

from coded_flows.types import PostgresDsn

# Direct validation of a Postgres connection string
try:
    pg_dsn = PostgresDsn("postgresql://user:password@localhost:5432/database")
    print(f"Valid Postgres DSN: {pg_dsn}")

    # Access components
    print(f"Host: {pg_dsn.hosts()}")
    print(f"Scheme: {pg_dsn.scheme}")
    print(f"Path (database): {pg_dsn.path}")
except ValueError as e:
    print(f"Invalid Postgres DSN: {e}")

CockroachDsn

Represents a CockroachDB connection string (DSN).

This type is a Pydantic type and can be used as a helper for validation and parsing of CockroachDB connection strings.

from coded_flows.types import CockroachDsn

# Direct validation of a CockroachDB connection string
try:
    cockroach_dsn = CockroachDsn("cockroachdb://user:password@localhost:26257/database")
    print(f"Valid CockroachDB DSN: {cockroach_dsn}")

    # Access components
    print(f"Host: {cockroach_dsn.host}")
    print(f"Port: {cockroach_dsn.port}")
    print(f"Database: {cockroach_dsn.path.lstrip('/')}")
except ValueError as e:
    print(f"Invalid CockroachDB DSN: {e}")

AmqpDsn

Represents an AMQP connection string (DSN).

This type is a Pydantic type and can be used as a helper for validation and parsing of AMQP connection strings.

from coded_flows.types import AmqpDsn

# Direct validation of an AMQP connection string
try:
    amqp_dsn = AmqpDsn("amqp://user:password@rabbitmq:5672/vhost")
    print(f"Valid AMQP DSN: {amqp_dsn}")

    # Access components
    print(f"Username: {amqp_dsn.username}")
    print(f"Host: {amqp_dsn.host}")
    print(f"Virtual host: {amqp_dsn.path.lstrip('/')}")
except ValueError as e:
    print(f"Invalid AMQP DSN: {e}")

RedisDsn

Represents a Redis connection string (DSN).

This type is a Pydantic type and can be used as a helper for validation and parsing of Redis connection strings.

from coded_flows.types import RedisDsn

# Direct validation of a Redis connection string
try:
    redis_dsn = RedisDsn("redis://user:password@redis:6379/0")
    print(f"Valid Redis DSN: {redis_dsn}")

    # Access components
    print(f"Host: {redis_dsn.host}")
    print(f"Port: {redis_dsn.port}")
    print(f"Database: {redis_dsn.path.lstrip('/')}")
except ValueError as e:
    print(f"Invalid Redis DSN: {e}")

MongoDsn

Represents a MongoDB connection string (DSN).

This type is a Pydantic type and can be used as a helper for validation and parsing of MongoDB connection strings.

from coded_flows.types import MongoDsn

# Direct validation of a MongoDB connection string
try:
    mongo_dsn = MongoDsn("mongodb://user:password@mongodb:27017/database")
    print(f"Valid MongoDB DSN: {mongo_dsn}")

    # Access components
    print(f"Host: {mongo_dsn.hosts()}")
    print(f"Database: {mongo_dsn.path.lstrip('/')}")
except ValueError as e:
    print(f"Invalid MongoDB DSN: {e}")

KafkaDsn

Represents a Kafka connection string (DSN).

This type is a Pydantic type and can be used as a helper for validation and parsing of Kafka connection strings.

from coded_flows.types import KafkaDsn

# Direct validation of a Kafka connection string
try:
    kafka_dsn = KafkaDsn("kafka://kafka-broker:9092/topic")
    print(f"Valid Kafka DSN: {kafka_dsn}")

    # Access components
    print(f"Host: {kafka_dsn.host}")
    print(f"Port: {kafka_dsn.port}")
    print(f"Topic: {kafka_dsn.path.lstrip('/')}")
except ValueError as e:
    print(f"Invalid Kafka DSN: {e}")

NatsDsn

Represents a NATS connection string (DSN).

This type is a Pydantic type and can be used as a helper for validation and parsing of NATS connection strings.

from coded_flows.types import NatsDsn

# Direct validation of a NATS connection string
try:
    nats_dsn = NatsDsn("nats://user:password@nats:4222")
    print(f"Valid NATS DSN: {nats_dsn}")

    # Access components
    print(f"Host: {nats_dsn.hosts()}")
except ValueError as e:
    print(f"Invalid NATS DSN: {e}")

MySQLDsn

Represents a MySQL connection string (DSN).

This type is a Pydantic type and can be used as a helper for validation and parsing of MySQL connection strings.

from coded_flows.types import MySQLDsn

# Direct validation of a MySQL connection string
try:
    mysql_dsn = MySQLDsn("mysql://user:password@mysql:3306/database")
    print(f"Valid MySQL DSN: {mysql_dsn}")

    # Access components
    print(f"Username: {mysql_dsn.username}")
    print(f"Host: {mysql_dsn.host}")
    print(f"Port: {mysql_dsn.port}")
    print(f"Database: {mysql_dsn.path.lstrip('/')}")
except ValueError as e:
    print(f"Invalid MySQL DSN: {e}")

MariaDBDsn

Represents a MariaDB connection string (DSN).

This type is a Pydantic type and can be used as a helper for validation and parsing of MariaDB connection strings.

from coded_flows.types import MariaDBDsn

# Direct validation of a MariaDB connection string
try:
    mariadb_dsn = MariaDBDsn("mariadb://user:password@mariadb:3306/database")
    print(f"Valid MariaDB DSN: {mariadb_dsn}")

    # Access components
    print(f"Username: {mariadb_dsn.username}")
    print(f"Host: {mariadb_dsn.host}")
    print(f"Port: {mariadb_dsn.port}")
    print(f"Database: {mariadb_dsn.path.lstrip('/')}")
except ValueError as e:
    print(f"Invalid MariaDB DSN: {e}")

MacAddress

Represents a MAC address.

EmailStr

Represents an email address string.

Bytes

Represents a bytes object.

Compatible types: Bytes, Base64Bytes

Bytearray

Represents a mutable array of bytes.

Base64Bytes

Represents base64-encoded bytes.

BytesIOType

Represents a BytesIO object for in-memory byte streams.

Path

Represents a file system path using the Python module pathlib.

Compatible types: Path, NewPath, FilePath, DirectoryPath

This type is a Pydantic type and can be used as a helper for validation and parsing of file system paths.

from coded_flows.types import Path

# Direct validation of a path string
try:
    path = Path("/home/user/documents")
    print(f"Valid path: {path}")
    print(f"Parent directory: {path.parent}")
except ValueError as e:
    print(f"Invalid path: {e}")

NewPath

Represents a new file system path (to be created) using the Python module pathlib..

This type is a Pydantic type and can be used as a helper for validation and parsing of new file system paths.

from coded_flows.types import NewPath

# Direct validation of a path that must not exist yet
try:
    new_path = NewPath("/home/user/documents/new_file.txt")
    print(f"Valid new path: {new_path}")

    # If the path already exists, it will raise a ValueError
    # This is useful for ensuring you don't overwrite existing files
except ValueError as e:
    print(f"Path already exists or invalid: {e}")

FilePath

Represents a path to an existing file using the Python module pathlib..

This type is a Pydantic type and can be used as a helper for validation and parsing of file paths.

from coded_flows.types import FilePath

# Direct validation of a path that must be an existing file
try:
    file_path = FilePath("/etc/hosts")
    print(f"Valid file path: {file_path}")
except ValueError as e:
    print(f"Not a valid file or file doesn't exist: {e}")

DirectoryPath

Represents a path to a directory using the Python module pathlib..

This type is a Pydantic type and can be used as a helper for validation and parsing of directory paths.

from coded_flows.types import DirectoryPath

# Direct validation of a path that must be an existing directory
try:
    dir_path = DirectoryPath("/home/user")
    print(f"Valid directory path: {dir_path}")
except ValueError as e:
    print(f"Not a valid directory or directory doesn't exist: {e}")

UUID

Represents a universally unique identifier.

Compatible types: UUID, UUID1, UUID3, UUID4, UUID5

UUID1

Represents a UUID version 1 (time-based).

UUID3

Represents a UUID version 3 (name-based with MD5).

UUID4

Represents a UUID version 4 (random).

UUID5

Represents a UUID version 5 (name-based with SHA-1).

JsonValue

Represents a value that can be serialized to JSON, such as lists, dictionaries, strings, numbers, booleans, or null.

Compatible types: List, DataRecords, Dict, DataDict, Str, Base64Str, CountryAlpha2, CountryAlpha3, CountryNumericCode, CountryShortName, EmailStr, Currency, Json, MacAddress, Bool, Int, PositiveInt, NegativeInt, Float, Longitude, Latitude, PositiveFloat, NegativeFloat, FiniteFloat, Null

Json

Represents a JSON string.

SecretStr

Represents a secret string, typically masked in output.

This type is a Pydantic type and can be used as a helper for validation and parsing of sensitive strings.

from coded_flows.types import SecretStr

# Direct validation of a sensitive string (like a password)
try:
    secret = SecretStr("my_secure_password")

    # The string representation will be masked
    print(f"Representation: {secret}")

    # To access the actual value (when needed)
    print(f"Length of secret: {len(secret.get_secret_value())}")

    # Use in secure context
    actual_password = secret.get_secret_value()
except ValueError as e:
    print(f"Error creating secret: {e}")

Color

Represents a color value (e.g., hex code or name).

This type is a Pydantic type and can be used as a helper for validation and parsing of color values.

from coded_flows.types import Color

# Direct validation of a color in various formats
try:
    # RGB hex format
    color1 = Color("#ff0000")
    print(f"Valid color from hex: {color1}")
    print(f"RGB: {color1.as_rgb()}")
    print(f"RGB tuple: {color1.as_rgb_tuple()}")

    # RGB tuple format
    color2 = Color((0, 255, 0))
    print(f"Valid color from tuple: {color2}")
    print(f"Hex: {color2.as_hex()}")

    # Color name
    color3 = Color("blue")
    print(f"Valid color from name: {color3}")
    print(f"HSL: {color3.as_hsl()}")
except ValueError as e:
    print(f"Invalid color: {e}")

Longitude

Represents a longitude coordinate.

Latitude

Represents a latitude coordinate.

Coordinate

Represents a geographic coordinate (latitude and longitude pair).

This type is a Pydantic type and can be used as a helper for validation and parsing of coordinates.

from coded_flows.types import Coordinate

# As a tuple (latitude, longitude)
coord1 = Coordinate(40.7128, -74.0060)  # New York City
print(f"Coordinate: {coord1}")

# Access individual values
print(f"Latitude: {coord1.latitude}, Longitude: {coord1.longitude}")

PILImage

Represents an image object from the PIL (Pillow) library.

MediaData

Represents media data, such as images or other binary data.

Compatible types: MediaData, NDArray, Bytes, BytesIOType