iporter 0.2.0

A pure tag system for parameter schemas in tools, pipelines, and agents

Iporter

A pure tag system for parameter schemas in tools, pipelines, and agents. Define structures, validate values, match providers, and auto-batch arrays—all with a concise, Zod-inspired API and zero dependencies.

Installation

pip install iporter

Quick Start

import iporter as p

# Define tags
user_tag = p.object({
    "name": p.string(min_length=1, max_length=100),
    "age": p.number().int().min(0).max(150),
    "email": p.string(),
    "role": p.enum(["admin", "user", "guest"]),
})

# Assign values
user_tag.assign({
    "name": "John Doe",
    "age": 30,
    "email": "[e-mail]",
    "role": "admin"
})

# Match tags
required = p.object({"name": p.string(), "age": p.number()})
provided = p.object({"name": p.string(), "age": p.number()})
result = p.match_tags(required, provided)
print(result.is_compatible)  # True

# Auto-batching with Porter
required = p.object({"x": p.number(), "y": p.string()})
provA = p.object({"x": p.array(p.number()), "y": p.string()})
provA.assign({"x": [1, 2, 3], "y": "hello"})

porter = p.Porter(required)
porter.add_provides(provA)
porter.assign()
print(required.expanded_values)
# [{"x": 1, "y": "hello"}, {"x": 2, "y": "hello"}, {"x": 3, "y": "hello"}]

Tag Types

Type	Description
StringTag	String values with length constraints
NumberTag	Numeric values (int/float) with min/max constraints
BooleanTag	Boolean values
NullTag	Null/None values
DateTimeTag	DateTime values
ArrayTag	Array/list values with item type constraints
ObjectTag	Object/dict values with field definitions
UnionTag	Union types (multiple possible types)
EnumTag	Enum values (restricted set of values)
OptionalTag	Optional values (allows None)
NullableTag	Nullable values (allows None)
LiteralTag	Literal values (exact match)
FileTag	File values with extension/size/MIME type constraints

Tag Assignment

tag = p.string()
tag.assign("hello")
print(tag.has_value)  # True
print(tag.value)      # "hello"

# Assignment validates type
tag.assign(123)  # Raises ValueError

Default Values

# Constructor syntax
tag = p.string(default="hello")
print(tag.value)      # "hello"
print(tag.has_value)  # True
print(tag.is_default) # True

# Chainable syntax
tag = p.string().default("hello")

# Explicit assign overrides default
tag.assign("world")
print(tag.is_default) # False

Custom Validation

# Add custom validation function
def check_non_empty(v):
    if len(v) == 0:
        raise ValueError("Cannot be empty")

tag = p.string().validate(check_non_empty)
tag.assign("hello")  # Passes
tag.assign("")       # Raises ValueError: Cannot be empty

# Chainable with other methods
tag = p.string().name("email").validate(lambda v: "@" in v or ValueError("Invalid email"))
tag.assign("[e-mail]")  # Passes

Tag Matching

required = p.object({"name": p.string(), "age": p.number()})
provided = p.object({"name": p.string(), "age": p.number(), "email": p.string()})

result = p.match_tags(required, provided)
print(result.is_compatible)  # True
print(result.extra_fields)   # ["email"]

Field Aliases

# Register aliases for fields with same meaning but different names
p.register_alias("email", "email_address")
p.register_alias("email", "user_email")

# Or batch register
p.register_aliases({
    "email": ["email_address", "user_email", "e_mail"],
    "name": ["username", "user_name"],
})

# Matching automatically considers aliases
required = p.object({"email": p.string()})
provided = p.object({"email_address": p.string()})

result = p.match_tags(required, provided)
print(result.is_compatible)  # True

# Resolve alias to canonical name
print(p.resolve_alias("email_address"))  # "email"
print(p.get_aliases("email"))  # ["email_address", "user_email"]

# Clear all aliases
p.clear_aliases()

Field Hierarchy

# Register is-a (subtype) relationships
p.register_subtype("pencil", "pen")      # pencil is-a pen
p.register_subtype("fountain_pen", "pen")  # fountain_pen is-a pen
p.register_subtypes("pencil", ["pen", "stationery"])  # Multiple inheritance

# Check hierarchy
p.is_subtype("pencil", "pen")       # True
p.is_subtype("pen", "pencil")       # False (directional!)

# Query
p.get_supertypes("pencil")          # ["pen"]
p.get_subtypes("pen")               # ["pencil", "fountain_pen"]

# Matching considers hierarchy
required = p.object({"pen": p.string()})
provided = p.object({"pencil": p.string()})
result = p.match_tags(required, provided)
print(result.is_compatible)  # True

Hierarchy-Based Aggregation

When a provider has multiple fields that are subtypes of the same required field, they are automatically collected:

p.register_subtype("read1", "read")
p.register_subtype("read2", "read")

prov = p.object({"read1": p.string(), "read2": p.string()})
prov.assign({"read1": "1.fq", "read2": "2.fq"})

# req expects array → aggregate into one array
req = p.object({"read": p.array(p.string())})
porter = p.Porter(req)
porter.add_provides(prov)
porter.assign()
print(req.value)  # {"read": ["1.fq", "2.fq"]}

# req expects scalar → expand into batch items
req = p.object({"read": p.string()})
porter = p.Porter(req)
porter.add_provides(prov)
porter.assign()
print(req.expanded_values)  # [{"read": "1.fq"}, {"read": "2.fq"}]

Porter Orchestration

# Define tags
required = p.object({"x": p.number(), "y": p.string()})
provA = p.object({"x": p.array(p.number()), "y": p.string()})
provB = p.object({"y": p.array(p.string())})

# Assign values to providers
provA.assign({"x": [1, 2], "y": "hello"})
provB.assign({"y": ["a", "b"]})

# Create Porter and configure
porter = p.Porter(required)
porter.add_provides(provA, provB, mode="product")

# Assign values from providers to required
porter.assign()
print(required.expanded_values)
# 6 items (Cartesian product)

Identity Tracking and produce()

Identity fields track which values correspond across batch items. When auto_batch expands arrays, the expanded fields are automatically marked as identity on the Tag.

# Provider with arrays that will be expanded
prov0 = p.object({
    "sample_id": p.array(p.string()),
    "input_file": p.array(p.file()),
})
prov0.assign({
    "sample_id": ["S1", "S2", "S3"],
    "input_file": ["/data/s1.bam", "/data/s2.bam", "/data/s3.bam"],
})

req = p.object({"sample_id": p.string(), "input_file": p.file()})
porter = p.Porter(req)
porter.add_provides(prov0, mode="zip")
porter.assign()

# After assign, expanded fields are auto-marked as identity
print(req.shape["sample_id"].is_identity)  # True

# Tool executes, user creates output ObjectTags
tool_outputs = []
for item in req.expanded_values:
    tag = p.object({"result_file": p.file()})
    tag.assign({"result_file": tool.run(item)})
    tool_outputs.append(tag)

# produce() inherits identity + merges tool outputs
prov1 = porter.produce(tool_outputs)
print(prov1.value)
# {
#   "sample_id": ["S1", "S2", "S3"],              # inherited from identity
#   "result_file": ["/out/s1.vcf", "/out/s2.vcf", "/out/s3.vcf"],  # from tool outputs
# }

# Identity map for downstream verification
print(prov1._identity_map)
# {0: {"sample_id": "S1"}, 1: {"sample_id": "S2"}, 2: {"sample_id": "S3"}}

Manual Identity Marking

You can also manually mark fields as identity:

req = p.object({
    "group": p.string().identity(),  # manually marked
    "data": p.number(),
})
print(req.shape["group"].is_identity)  # True

File Tags

# Basic file tag
tag = p.file()
tag.validate_type("/path/to/file.txt")  # True

# With constraints
tag = p.file(
    extensions=[".json", ".csv"],
    max_size=1024 * 1024,  # 1MB
    mime_types=["application/json", "text/csv"],
)

# Validate file path
tag.validate_type("/path/to/data.json")  # True
tag.validate_type("/path/to/image.png")  # False

API Reference

Tag Builders

Builder	Description
p.string(min_length, max_length, name, description, default)	String tag
p.number(is_int, min_val, max_val, name, description, default)	Number tag
p.boolean(name, description, default)	Boolean tag
p.null(name, description)	Null tag
p.datetime(name, description, default)	DateTime tag
p.array(item_tag, name, description, default)	Array tag
p.object(shape, name, description, default)	Object tag
p.union(tags, name, description)	Union tag
p.enum(values, name, description, default)	Enum tag
p.optional(inner_tag, name, description)	Optional tag
p.nullable(inner_tag, name, description)	Nullable tag
p.literal(value, name, description)	Literal tag
p.file(extensions, max_size, mime_types, name, description)	File tag

Tag Methods

Method	Description
tag.assign(value)	Assign a value (validates type + custom validator)
tag.validate_type(value)	Check if value matches tag type
tag.type_name()	Get type name string
tag.name(value)	Set tag name (chainable)
tag.description(desc)	Set description (chainable)
tag.default(value)	Set default value (chainable)
tag.validate(fn)	Set custom validation function (chainable)
tag.get_info()	Get name and description as dict
tag.value	Get assigned value (or default)
tag.has_value	Check if value assigned (or has default)
tag.is_default	Check if current value is from default

ObjectTag Properties

Property	Description
tag.shape	Get field definitions
tag.required_fields	Get required field names
tag.optional_fields	Get optional field names
tag.expanded_values	Get expanded values (after Porter batching)

Porter

Method/Property	Description
p.Porter(required)	Create Porter with required tag
porter.add_provides(*providers, mode=None, dup_strategy=None)	Add provider tags
porter.assign()	Assign values from providers to required tag
porter.produce(tool_outputs)	Create output provider inheriting identity from batch items
porter.required	Get the required tag
porter.providers	Get provider tags dict
porter.batch_mode	Get/set batch mode: "zip" (default) or "product"
porter.dup_strategy	Get/set duplicate key strategy: "merge" (default), "first", "last", or "error"

Tag Methods (Extended)

Method	Description
tag.identity()	Mark tag as identity field (chainable)
tag.is_identity	Check if tag is an identity field

Matching and Batching

Function	Description
p.match_tags(required, provided)	Match two tags
p.match_tags_multi(required, provided_list)	Match against multiple providers
p.auto_batch(required, providers, values, mode, dup_strategy)	Auto-batch from providers

Aliases

Function	Description
p.register_alias(canonical, alias)	Register a field alias
p.register_aliases(alias_map)	Register multiple aliases
p.resolve_alias(name)	Resolve alias to canonical name
p.get_aliases(canonical)	Get all aliases for a name
p.clear_aliases()	Clear all aliases

Hierarchy

Function	Description
p.register_subtype(subtype, supertype)	Register a single is-a relationship
p.register_subtypes(subtype, supertypes)	Register multiple supertypes
p.register_hierarchy(hierarchy_map)	Bulk register from dict
p.is_subtype(subtype, supertype)	Check transitive is-a
p.get_supertypes(subtype)	Get direct supertypes
p.get_subtypes(supertype)	Get direct subtypes
p.clear_hierarchy()	Clear all relationships

Architecture

See ARCHITECTURE.md for detailed architecture documentation.

License

MIT