i2.castgraph¶

A lightweight transformation service for Python that solves the “stable role, unstable representation” problem: a resource has a consistent semantic role (e.g., configuration, text, structured record) but appears in many forms (filepath, string, dict, custom class), while consumers expect specific representations. castgraph organizes transformations as a graph of “kinds” (data representations) and routes requests through the best available path.

Key concepts

Kind: Any hashable identifier for a data representation (type, string, custom marker)
Transformation: An edge in the graph that converts one kind to another
Kind Predicate (isa): A function that determines if an object is of a kind
TransformationGraph: The main registry with graph-oriented interface

Solution patterns

Type Converter / Conversion Service: central registry mapping (FromKind, ToKind) to transformer functions.
Adapter: each edge adapts one representation to another.
Strategy: routing/selection among multiple possible transformations via cost/priority.
(Optional) Canonical Data Model: a hub kind to reduce pairwise conversions.
DDD Anti-Corruption Layer (ACL): keep external formats outside the core domain.
Typeclass / Multimethod idiom: dispatch based on (source kind, target kind).

Minimal example (new kind-based interface)

Use the new TransformationGraph with flexible kinds (not limited to types).

>>> from i2.castgraph import TransformationGraph
>>> graph = TransformationGraph()
>>> # Add nodes with predicates
>>> graph.add_node('text', isa=lambda x: isinstance(x, str))
>>> graph.add_node('json_dict', isa=lambda x: isinstance(x, dict))
>>> # Add transformation edges
>>> @graph.register_edge('text', 'json_dict')
... def text_to_json(t, ctx):
...     import json
...     return json.loads(t or "{}")
>>> # Transform using kinds (need explicit from_kind since 'text' != str)
>>> result = graph.transform('{"x": 1}', 'json_dict', from_kind='text')
>>> result["x"]
1

Legacy example (type-based interface)

The old ConversionRegistry interface still works but is deprecated.

>>> from i2.castgraph import ConversionRegistry
>>> import warnings
>>> class Path(str): ...
>>> class Text(str): ...
>>> class Record(dict): ...
>>> reg = ConversionRegistry()
>>> with warnings.catch_warnings():
...     warnings.simplefilter("ignore")
...     @reg.register(Path, Text)
...     def path_to_text(p, ctx):
...         fs = (ctx or {}).get("fs", {})
...         return Text(fs.get(str(p), ""))
...     @reg.register(Text, Record, cost=0.5)
...     def text_to_record(t, ctx):
...         import json
...         return Record(json.loads(t or "{}"))
>>> ctx = {"fs": {"/app/data.json": '{"x": 1}'}}
>>> with warnings.catch_warnings():
...     warnings.simplefilter("ignore")
...     out = reg.convert(Path("/app/data.json"), Record, context=ctx)
>>> isinstance(out, Record) and out["x"] == 1
True

Main tools

TransformationGraph: the main graph-based registry (recommended).
- .add_node(kind, isa=None): add a kind with optional predicate.
- .add_edge(src, dst, func, cost=1.0): add a transformation edge.
- .register_edge(src, dst, cost=1.0): decorator to add an edge.
- .transform(obj, to_kind, from_kind=None, context=None): transform with multi-hop routing.
- .transform_any(obj, to_kind, context=None): transform with automatic kind detection.
- .get_transformer(from_kind, to_kind): get a composed transformer function.
- .detect_kind(obj): detect the kind of an object.
- .reachable_from(kind): get all reachable kinds.
- .sources_for(kind): get all source kinds.
- .kinds(): get all registered kinds.
ConversionRegistry: DEPRECATED - use TransformationGraph instead.
- .register(From, To, cost=1.0): DEPRECATED - use .register_edge() instead.
- .convert(obj, ToType, context=None): DEPRECATED - use .transform() instead.
Kind: Optional wrapper for explicit kind specification with predicates.
KindMatch: Truthy result from kind predicates that can carry metadata.
ConversionError: raised when no route exists between kinds.

Design guidelines

Define a single TransformationGraph per bounded context; keep edges local.
Prefer small, testable transformer functions with explicit kinds.
Use a canonical domain kind as a hub when many formats interoperate.
Assign costs to prefer fast/accurate routes; tune with metrics.
Pass a context dict for side-channel knobs (I/O, flags, cache handles).
Cache paths (via lru_cache) and consider result caching for hot transformations.
Keep adapters at the boundaries; the core domain should consume domain kinds.
Add identity edges implicitly; avoid no-op boilerplate.
Write doctests on each transformer to lock behavior and invariants.
Use bare hashables (types, strings) as kinds; Kind class is optional.

Migration guide

Old code using ConversionRegistry:

reg = ConversionRegistry()
@reg.register(SrcType, DstType)
def convert_func(obj, ctx): ...
result = reg.convert(obj, DstType)

New code using TransformationGraph:

graph = TransformationGraph()
@graph.register_edge(SrcType, DstType)
def transform_func(obj, ctx): ...
result = graph.transform(obj, DstType)

Or with string kinds:

graph = TransformationGraph()
graph.add_node('src_format', isa=lambda x: ...)
@graph.register_edge('src_format', 'dst_format')
def transform_func(obj, ctx): ...
result = graph.transform(obj, 'dst_format')

Design heritage

castgraph is a composition of well-known patterns centered on a **Type Converter / Conversion Service**, with Adapter edges and Strategy-based route selection. At system boundaries, it complements DDD’s Anti-Corruption Layer and can employ an integration Canonical Data Model to curb O(n²) pairwise mappings. Its (FromType, ToType) dispatch style mirrors typeclass/multimethod idioms. For background reading, see:

.NET TypeConverter: https://learn.microsoft.com/dotnet/api/system.componentmodel.typeconverter
Spring ConversionService: https://docs.spring.io/spring-framework/reference/core/validation/convert.html
Apache Camel Type Converter: https://camel.apache.org/manual/type-converter.html
Adapter: https://refactoring.guru/design-patterns/adapter
Strategy: https://refactoring.guru/design-patterns/strategy
Anti-Corruption Layer: https://martinfowler.com/bliki/AntiCorruptionLayer.html
Canonical Data Model: https://www.enterpriseintegrationpatterns.com/patterns/messaging/CanonicalDataModel.html
PEP 443 singledispatch: https://peps.python.org/pep-0443/

Related

Issue that sparked this implementation: https://github.com/i2mint/i2/issues/79
Computational path resolution: https://github.com/i2mint/meshed/discussions/71
Subsuming concept - “routing”: https://github.com/i2mint/i2/discussions/68

Functions

design_guidelines()

Returns concise guidance for organizing casting in Python.

Classes

`ConversionRegistry`()	DEPRECATED: Use TransformationGraph instead.
`Edge`(src, dst, func[, cost])	DEPRECATED: Use Transformation instead.
`Kind`(identifier[, isa])	Optional marker for explicit kind specification.
`KindMatch`([metadata])	Result of a successful kind predicate match.
`Transformation`(src, dst, func[, cost])	An edge in the transformation graph.
`TransformationGraph`()	A graph-based registry of transformations between kinds (data representations).

Exceptions

ConversionError

exception i2.castgraph.ConversionError[source]¶: Bases: TypeError

class i2.castgraph.ConversionRegistry[source]¶

Bases: object

DEPRECATED: Use TransformationGraph instead.

A graph-based registry of converters between Python types with:

registration decorator

shortest-path (by total cost) routing

MRO-aware fallback for source types

caching of paths and (optionally) results

Design notes:

Each converter has signature: func(obj, context) -> converted_obj
Identity edges are implicit (T -> T) with cost 0.
If multiple routes exist, the minimum total cost path is chosen.

convert(obj, to_type, *, context=None, use_result_cache=False)[source]¶

Convert obj to to_type, possibly via multi-hop.

Parameters:

obj (Any) – Source object to convert.
to_type (type[TypeVar(U)]) – Desired target type.
context (dict | None) – Arbitrary context propagated through the chain (e.g., config, flags).
use_result_cache (bool) – If True, cache results keyed by (id(obj), to_type).

Returns:

Converted object.

Return type:

TypeVar(U)

Raises:

ConversionError – If no conversion path is found.

Examples

>>> reg = ConversionRegistry()
>>> class X: ...
>>> class Y: ...
>>> class Z: ...
>>> @reg.register(X, Y)
... def x_to_y(x, ctx): return Y()
...
>>> @reg.register(Y, Z)
... def y_to_z(y, ctx): return Z()
...
>>> isinstance(reg.convert(X(), Z), Z)
True

MRO fallback: if a converter is registered for a base class, it applies to a subclass.

>>> class Base: ...
>>> class Sub(Base): ...
>>> class Out: ...
>>> reg2 = ConversionRegistry()
>>> @reg2.register(Base, Out)
... def base_to_out(b, ctx): return Out()
...
>>> isinstance(reg2.convert(Sub(), Out), Out)
True

register(src=None, dst=None, *, cost=1.0)[source]¶

Decorator to register a converter function.

Return type:: Callable[[Callable[[Any, Optional[dict]], Any]], Callable[[Any, Optional[dict]], Any]]

>>> reg = ConversionRegistry()
>>> class A: ...
>>> class B: ...
>>> @reg.register(A, B)
... def a_to_b(a, ctx): return B()
...
>>> isinstance(reg.convert(A(), B), B)
True

Types can be inferred from annotations:

>>> class X: ...
>>> class Y: ...
>>> @reg.register()
... def x_to_y(x: X, ctx) -> Y:
...     return Y()
>>> isinstance(reg.convert(X(), Y), Y)
True

class i2.castgraph.Edge(src, dst, func, cost=1.0)[source]¶

Bases: object

DEPRECATED: Use Transformation instead. Kept for backward compatibility.

class i2.castgraph.Kind(identifier, isa=None)[source]¶

Bases: object

Optional marker for explicit kind specification.

A Kind wraps a hashable identifier and optionally an ‘isa’ predicate. Users are NOT required to use this class - bare hashables work fine. This class is for when you want to be explicit or bundle identifier + predicate.

>>> text_kind = Kind('text', isa=lambda x: isinstance(x, str))
>>> text_kind.identifier
'text'
>>> text_kind.isa("hello")
True

isa(obj)[source]¶

Check if obj is of this kind (predicate/recognizer function).

Return type:: bool | KindMatch

class i2.castgraph.KindMatch(metadata=None)[source]¶

Bases: object

Result of a successful kind predicate match.

Evaluates to True but can carry additional metadata about the match that downstream transformations might use.

>>> match = KindMatch({'encoding': 'utf-8', 'analyzed': True})
>>> bool(match)
True
>>> match.metadata
{'encoding': 'utf-8', 'analyzed': True}

class i2.castgraph.Transformation(src, dst, func, cost=1.0)[source]¶

Bases: object

An edge in the transformation graph.

Represents a transformation function from one kind to another.

class i2.castgraph.TransformationGraph[source]¶

Bases: object

A graph-based registry of transformations between kinds (data representations).

A “kind” is any hashable identifier for a data representation - it can be a type, a string, or any custom marker. The graph supports:

Flexible kind system (not limited to Python types)

Graph-oriented interface (add_node, add_edge)

Pluggable kind detection via predicates

Shortest-path (by total cost) routing

MRO-aware fallback for type-based kinds

Caching of paths and (optionally) results

Design notes:

Each transformer has signature: func(obj, context) -> transformed_obj
Identity edges are implicit (K -> K) with cost 0
If multiple routes exist, the minimum total cost path is chosen
Kinds can be types, strings, or any hashable objects

add_edge(src, dst, func, *, cost=1.0)[source]¶

Add a transformation (edge) between two kinds.

Automatically adds nodes if they don’t exist.

Parameters:

src (Hashable | Kind) – Source kind
dst (Hashable | Kind) – Destination kind
func (Callable) – Transformation function with signature func(obj, context) -> transformed_obj or func(obj) -> transformed_obj (will be wrapped)
cost (float) – Cost of this transformation (lower is preferred)

Return type:

None

Examples

>>> graph = TransformationGraph()
>>> def text_to_int(s, ctx): return int(s)
>>> graph.add_edge('text', int, text_to_int)

add_node(kind, isa=None)[source]¶

Add a kind (node) to the graph with optional predicate.

Parameters:

kind (Hashable | Kind) – The kind identifier (can be a type, string, or Kind object)
isa (Callable[[Any], bool | KindMatch] | None) – Optional predicate function to detect if an object is of this kind

Return type:

None

Examples

>>> graph = TransformationGraph()
>>> graph.add_node('text', isa=lambda x: isinstance(x, str))
>>> graph.add_node(int)  # Type implies isinstance check

convert(obj, to_type, *, context=None, use_result_cache=False)[source]¶

DEPRECATED: Use transform() instead.

This method is kept for backward compatibility.

Return type:: TypeVar(U)

Examples

>>> import warnings
>>> graph = TransformationGraph()
>>> @graph.register_edge(str, int)
... def str_to_int(s, ctx): return int(s)
>>> with warnings.catch_warnings():
...     warnings.simplefilter("ignore")
...     result = graph.convert("42", int)
>>> result
42

detect_kind(obj)[source]¶

Detect the kind of an object.

Uses custom detector if set, otherwise tries registered predicates in order. Returns None if no kind matches.

Parameters:: obj (Any) – Object to classify
Returns:: The detected kind identifier, or None if no match
Return type:: Hashable | None

Examples

>>> graph = TransformationGraph()
>>> graph.add_node('text', isa=lambda x: isinstance(x, str))
>>> graph.detect_kind("hello")
'text'

get_transformer(from_kind, to_kind, *, context=None)[source]¶

Get a function that transforms from_kind → to_kind.

Returns a composed transformer function (Pipe-like).

Parameters:

from_kind (Hashable | Kind) – Source kind
to_kind (Hashable | Kind) – Destination kind
context (dict | None) – Optional context to bake into the transformer

Returns:

A function that transforms objects from from_kind to to_kind

Return type:

Callable[[Any], Any]

Examples

>>> graph = TransformationGraph()
>>> @graph.register_edge(str, int)
... def str_to_int(s, ctx): return int(s)
>>> transformer = graph.get_transformer(str, int)
>>> transformer("42")
42

property ingress¶

Return decorator factory with attribute access for kinds.

This property provides a flexible interface for decorating functions to automatically transform their arguments to specified kinds.

Usage patterns:

Specify kind and argument name: @graph.ingress(‘text’, ‘content’) def func(content): …
Specify kind only (transforms first argument): @graph.ingress(‘text’) def func(arg): …
Use keyword argument: @graph.ingress(arg_name=’text’) def func(arg_name): …
Attribute-based syntax for registered kinds: @graph.ingress.text(‘content’) def func(content): …
Attribute-based for first argument: @graph.ingress.text def func(arg): …

Examples

>>> graph = TransformationGraph()
>>> graph.add_node('text', isa=lambda x: isinstance(x, str))
>>> graph.add_node(int)
>>> @graph.register_edge('text', int)
... def text_to_int(s, ctx): return int(s)
>>> @graph.ingress('text')
... def process(x):
...     return x + ' processed'
>>> # Can now pass int, will be transformed to text first

kinds()[source]¶

Get all registered kinds (nodes in the graph).

Returns:: Set of all registered kind identifiers
Return type:: set[Hashable]

Examples

>>> graph = TransformationGraph()
>>> graph.add_node('text')
>>> graph.add_node(int)
>>> 'text' in graph.kinds()
True

reachable_from(kind)[source]¶

Get all kinds reachable from this kind via transformations.

Parameters:: kind (Hashable | Kind) – The starting kind
Returns:: Set of all reachable kind identifiers
Return type:: set[Hashable]

Examples

>>> graph = TransformationGraph()
>>> # ... register transformations ...
>>> reachable = graph.reachable_from('text')

register(src=None, dst=None, *, cost=1.0)[source]¶

DEPRECATED: Use register_edge() instead.

This method is kept for backward compatibility.

Return type:: Callable[[Callable[[Any, Optional[dict]], Any]], Callable[[Any, Optional[dict]], Any]]

Examples

>>> import warnings
>>> graph = TransformationGraph()
>>> with warnings.catch_warnings():
...     warnings.simplefilter("ignore")
...     @graph.register(str, int)
...     def str_to_int(s, ctx): return int(s)

register_edge(src=None, dst=None, *, cost=1.0)[source]¶

Decorator to register a transformation edge.

Can infer src/dst from function annotations if not provided.

Parameters:

src (Hashable | Kind | None) – Source kind (inferred from annotations if None)
dst (Hashable | Kind | None) – Destination kind (inferred from annotations if None)
cost (float) – Cost of this transformation

Return type:

Callable

Examples

>>> graph = TransformationGraph()
>>> @graph.register_edge('text', int)
... def text_to_int(s, ctx): return int(s)

set_kind_detector(detector)[source]¶

Set a custom kind detector function.

The detector receives an object and returns a kind identifier or None.

Parameters:: detector (Callable[[Any], Hashable | None]) – Function that takes an object and returns its kind or None
Return type:: None

Examples

>>> graph = TransformationGraph()
>>> def my_detector(obj):
...     if isinstance(obj, str) and obj.startswith('{"'):
...         return 'json_string'
...     return None
>>> graph.set_kind_detector(my_detector)

sources_for(kind)[source]¶

Get all kinds that can be transformed to this kind.

Parameters:: kind (Hashable | Kind) – The destination kind
Returns:: Set of all source kind identifiers
Return type:: set[Hashable]

Examples

>>> graph = TransformationGraph()
>>> # ... register transformations ...
>>> sources = graph.sources_for(int)

transform(obj, to_kind, *, from_kind=None, context=None, use_result_cache=False)[source]¶

Transform obj to to_kind.

If from_kind not specified, uses type(obj) with MRO fallback.

Parameters:

obj (Any) – Object to transform
to_kind (Hashable | Kind) – Destination kind
from_kind (Hashable | Kind | None) – Source kind (inferred if None)
context (dict | None) – Optional context passed to transformation functions
use_result_cache (bool) – If True, cache results keyed by (id(obj), to_kind)

Returns:

Transformed object

Return type:

Any

Raises:

ConversionError – If no transformation path is found

Examples

>>> graph = TransformationGraph()
>>> @graph.register_edge(str, int)
... def str_to_int(s, ctx): return int(s)
>>> graph.transform("42", int)
42

transform_any(obj, to_kind, *, context=None, use_result_cache=False)[source]¶

Transform obj to to_kind with automatic kind detection.

Uses configured kind detector or fallback detection strategy.

Parameters:

obj (Any) – Object to transform
to_kind (Hashable | Kind) – Destination kind
context (dict | None) – Optional context passed to transformation functions
use_result_cache (bool) – If True, cache results

Returns:

Transformed object

Return type:

Any

Raises:

ConversionError – If no transformation path is found or kind cannot be detected

Examples

>>> graph = TransformationGraph()
>>> graph.add_node('text', isa=lambda x: isinstance(x, str))
>>> @graph.register_edge('text', int)
... def text_to_int(s, ctx): return int(s)
>>> graph.transform_any("42", int)
42

i2.castgraph.design_guidelines()[source]¶

Returns concise guidance for organizing casting in Python.

Return type:: str

>>> "registry" in design_guidelines().lower()
True