graflo.util.transform

Data transformation utilities for graph operations.

This module provides utility functions for transforming and standardizing data in various formats, particularly for graph database operations. It includes functions for date parsing, string standardization, and data cleaning.

Key Functions

• standardize: Standardize string keys and names
• parse_date_*: Various date parsing functions for different formats
• cast_ibes_analyst: Parse and standardize analyst names
• clear_first_level_nones: Clean dictionaries by removing None values
• parse_multi_item: Parse complex multi-item strings
• pick_unique_dict: Remove duplicate structures by content hash
• affix_gated_key: Admit a value only when it carries its marker affixes

Example

name = standardize("John. Doe, Smith") date = parse_date_standard("2023-01-01") analyst = cast_ibes_analyst("ADKINS/NARRA")

Attributes

ORDINAL_SUFFIX = ['st', 'nd', 'rd', 'th'] module-attribute

T = TypeVar('T') module-attribute

logger = logging.getLogger(__name__) module-attribute

Functions:

affix_gated_key(value, *, prefix='', suffix='', casefold=True, strip_chars=None)

Return value stripped of its marker affixes, or None if unmarked.

The affixes are the admission test, not a best-effort cleanup: a value carrying them is stripped and accepted as canonical key material, and a value missing either one yields None. Contrast :func:gated_normalized_key, which gates on a sibling field and whose strip_prefix is a silent no-op when absent — there, marked and unmarked values normalize to the same key and fuse.

Both affixes are required when given: a marker can be a prefix (ext_), a suffix (-legacy), or the pair that brackets a key. Each defaults to "", which every string carries, so naming one leaves the other unconstrained and naming neither admits everything while stripping nothing — the convention that lets a source participating unconditionally reuse the same function, and thus the same normal form, as one that is filtered.

None is a fall-through, not a drop. It is an empty value to identity digests, so an identity-funnel branch listing the output field is skipped and the record lands on its side-local branch — still ingested, just not into the cross-source cluster.

The marker test is case-sensitive even when casefold is set: casefolding applies to the surviving key, after the affixes have been removed.

Parameters:

Name	Type	Description	Default
value	str | None	Raw key material, expected to carry the affixes.	required
prefix	str	Leading marker admitting value; removed from it.	''
suffix	str	Trailing marker admitting value; removed from it.	''
casefold	bool	Casefold the surviving key.	True
strip_chars	str | None	Characters stripped from both ends of value before the affixes are tested (None strips whitespace).	None

Returns:

Type	Description
str | None	The normalized key, or None when value is missing, lacks either
str | None	affix, or is empty once both are removed.

Source code in graflo/util/transform.py

def affix_gated_key(
    value: str | None,
    *,
    prefix: str = "",
    suffix: str = "",
    casefold: bool = True,
    strip_chars: str | None = None,
) -> str | None:
    """Return *value* stripped of its marker affixes, or ``None`` if unmarked.

    The affixes are the admission test, not a best-effort cleanup: a value
    carrying them is stripped and accepted as canonical key material, and a
    value missing either one yields ``None``. Contrast
    :func:`gated_normalized_key`, which gates on a *sibling* field and whose
    ``strip_prefix`` is a silent no-op when absent — there, marked and unmarked
    values normalize to the same key and fuse.

    Both affixes are required when given: a marker can be a prefix (``ext_``), a
    suffix (``-legacy``), or the pair that brackets a key. Each defaults to
    ``""``, which every string carries, so naming one leaves the other
    unconstrained and naming neither admits everything while stripping nothing —
    the convention that lets a source participating unconditionally reuse the
    same function, and thus the same normal form, as one that is filtered.

    ``None`` is a fall-through, not a drop. It is an empty value to identity
    digests, so an identity-funnel branch listing the output field is skipped
    and the record lands on its side-local branch — still ingested, just not
    into the cross-source cluster.

    The marker test is case-sensitive even when *casefold* is set: casefolding
    applies to the surviving key, after the affixes have been removed.

    Args:
        value: Raw key material, expected to carry the affixes.
        prefix: Leading marker admitting *value*; removed from it.
        suffix: Trailing marker admitting *value*; removed from it.
        casefold: Casefold the surviving key.
        strip_chars: Characters stripped from both ends of *value* before the
            affixes are tested (``None`` strips whitespace).

    Returns:
        The normalized key, or ``None`` when *value* is missing, lacks either
        affix, or is empty once both are removed.
    """
    if value is None:
        return None
    key = str(value).strip(strip_chars)
    if not (key.startswith(prefix) and key.endswith(suffix)):
        return None
    # Overlapping affixes both "match" a short value without bracketing it:
    # ``"ABCX"`` starts with ``"ABC"`` and ends with ``"BCX"``, and removing
    # both would consume characters twice.
    if len(prefix) + len(suffix) > len(key):
        return None
    key = key[len(prefix) : len(key) - len(suffix)]
    if casefold:
        key = key.casefold()
    return key or None

camel_to_snake(s) cached

Convert camelCase/PascalCase names to snake_case.

Source code in graflo/util/transform.py

@lru_cache(maxsize=32768)
def camel_to_snake(s: str) -> str:
    """Convert camelCase/PascalCase names to snake_case."""
    if not s:
        return s
    step1 = _CAMEL_TO_SNAKE_STEP1_RE.sub(r"\1_\2", s)
    step2 = _CAMEL_TO_SNAKE_STEP2_RE.sub(r"\1_\2", step1)
    return step2.lower()

cast_ibes_analyst(s)

Splits and normalizes analyst name strings.

Handles various name formats like 'ADKINS/NARRA' or 'ARFSTROM J'.

Parameters:

Name	Type	Description	Default
s	str	Analyst name string.	required

Returns:

Name	Type	Description
tuple	tuple[str, str]	(last_name, first_initial)

Examples:

>>> cast_ibes_analyst('ADKINS/NARRA')
('ADKINS', 'N')
>>> cast_ibes_analyst('ARFSTROM      J')
('ARFSTROM', 'J')

Source code in graflo/util/transform.py

def cast_ibes_analyst(s: str) -> tuple[str, str]:
    """Splits and normalizes analyst name strings.

    Handles various name formats like 'ADKINS/NARRA' or 'ARFSTROM      J'.

    Args:
        s (str): Analyst name string.

    Returns:
        tuple: (last_name, first_initial)

    Examples:
        >>> cast_ibes_analyst('ADKINS/NARRA')
        ('ADKINS', 'N')
        >>> cast_ibes_analyst('ARFSTROM      J')
        ('ARFSTROM', 'J')
    """
    if " " in s or "\t" in s:
        r = s.split()[:2]
        if len(r) < 2:
            return r[0], ""
        else:
            return r[0], r[1][:1]
    else:
        r = s.split("/")
        if s.startswith("/"):
            r = r[1:3]
        else:
            r = r[:2]
        if len(r) < 2:
            return r[0], ""
        else:
            return r[0], r[1][:1]

clear_first_level_nones(docs, keys_keep_nones=None)

Removes None values from dictionaries, with optional key exceptions.

Parameters:

Name	Type	Description	Default
docs	list	List of dictionaries to clean.	required
keys_keep_nones	list	Keys to keep even if their value is None.	None

Returns:

Name	Type	Description
list	list[dict]	Cleaned list of dictionaries.

Example

docs = [{"a": 1, "b": None}, {"a": None, "b": 2}] clear_first_level_nones(docs, keys_keep_nones=["a"]) [{"a": 1}, {"a": None, "b": 2}]

Source code in graflo/util/transform.py

def clear_first_level_nones(
    docs: list[dict], keys_keep_nones: list | None = None
) -> list[dict]:
    """Removes None values from dictionaries, with optional key exceptions.

    Args:
        docs (list): List of dictionaries to clean.
        keys_keep_nones (list, optional): Keys to keep even if their value is None.

    Returns:
        list: Cleaned list of dictionaries.

    Example:
        >>> docs = [{"a": 1, "b": None}, {"a": None, "b": 2}]
        >>> clear_first_level_nones(docs, keys_keep_nones=["a"])
        [{"a": 1}, {"a": None, "b": 2}]
    """
    if keys_keep_nones is not None:
        docs = [
            {k: v for k, v in tdict.items() if v or k in keys_keep_nones}
            for tdict in docs
        ]
    return docs

coalesce_fields(doc, *, fields)

First non-empty value among fields on doc, or None.

The branch selector for the column-presence form of a routed derivation. When one resource derives a canonical attribute several ways — one per class its vertex_router collapses onto the aligned class, each keying from its own column — each derivation writes its own scratch field and returns None for the branches it does not serve. This picks the one that fired. (A derivation keyed by member needs none of this: its step carries a when guard and writes the attribute directly.)

A single writer per canonical attribute is the point. Two steps writing the same key work on a plain vertex step, whose buffer extraction skips None, but not behind a vertex_router: the router merges the buffer into one observation dict, where a later None overwrites an earlier real value.

Called with strategy: all, so a branch whose own columns are absent from the document skips without taking the coalesce down with it.

Parameters:

Name	Type	Description	Default
doc	dict[str, Any]	The merged observation.	required
fields	list[str]	Scratch field names, in priority order.	required

Returns:

Type	Description
Any	The first present, non-empty value, or None when none fired.

Source code in graflo/util/transform.py

def coalesce_fields(doc: dict[str, Any], *, fields: list[str]) -> Any:
    """First non-empty value among *fields* on *doc*, or ``None``.

    The branch selector for the column-presence form of a routed derivation.
    When one resource derives a canonical attribute several ways — one per
    class its ``vertex_router`` collapses onto the aligned class, each keying
    from its own column — each derivation writes its own scratch field and
    returns ``None`` for the branches it does not serve. This picks the one
    that fired. (A derivation keyed by *member* needs none of this: its step
    carries a ``when`` guard and writes the attribute directly.)

    A single writer per canonical attribute is the point. Two steps writing the
    same key work on a plain ``vertex`` step, whose buffer extraction skips
    ``None``, but not behind a ``vertex_router``: the router merges the buffer
    into one observation dict, where a later ``None`` overwrites an earlier
    real value.

    Called with ``strategy: all``, so a branch whose own columns are absent
    from the document skips without taking the coalesce down with it.

    Args:
        doc: The merged observation.
        fields: Scratch field names, in priority order.

    Returns:
        The first present, non-empty value, or ``None`` when none fired.
    """
    for field in fields:
        value = doc.get(field)
        if value is None:
            continue
        if isinstance(value, str) and not value.strip():
            continue
        return value
    return None

gated_normalized_key(gate, value, *, prefix, strip_prefix=None, casefold=True, strip_chars=None)

Return the normalized value when gate starts with prefix, else None.

Designed for conditional entity equivalence: emit a shared match key only for records that participate in it. None is treated as an empty value by identity digests, so an identity-funnel branch listing the output field is skipped and the record falls through to its side-local branch.

An empty prefix makes the gate always pass, which lets the other side of an equivalence reuse the same function (and thus the same normal form) unconditionally.

Parameters:

Name	Type	Description	Default
gate	str | None	Field deciding participation (e.g. u_number).	required
value	str | None	Raw key material to normalize.	required
prefix	str	Required prefix of gate; "" always passes.	required
strip_prefix	str | None	Prefix removed from value when present.	None
casefold	bool	Casefold the normalized value.	True
strip_chars	str | None	Characters stripped from both ends of value (None strips whitespace).	None

Returns:

Type	Description
str | None	The normalized key, or None when gate or value is missing or
str | None	the gate does not match.

Source code in graflo/util/transform.py

def gated_normalized_key(
    gate: str | None,
    value: str | None,
    *,
    prefix: str,
    strip_prefix: str | None = None,
    casefold: bool = True,
    strip_chars: str | None = None,
) -> str | None:
    """Return the normalized *value* when *gate* starts with *prefix*, else ``None``.

    Designed for conditional entity equivalence: emit a shared match key only
    for records that participate in it. ``None`` is treated as an empty value
    by identity digests, so an identity-funnel branch listing the output field
    is skipped and the record falls through to its side-local branch.

    An empty *prefix* makes the gate always pass, which lets the other side of
    an equivalence reuse the same function (and thus the same normal form)
    unconditionally.

    Args:
        gate: Field deciding participation (e.g. ``u_number``).
        value: Raw key material to normalize.
        prefix: Required prefix of *gate*; ``""`` always passes.
        strip_prefix: Prefix removed from *value* when present.
        casefold: Casefold the normalized value.
        strip_chars: Characters stripped from both ends of *value*
            (``None`` strips whitespace).

    Returns:
        The normalized key, or ``None`` when *gate* or *value* is missing or
        the gate does not match.
    """
    if value is None or gate is None:
        return None
    if not str(gate).startswith(prefix):
        return None
    key = str(value).strip(strip_chars)
    if strip_prefix:
        key = key.removeprefix(strip_prefix)
    if casefold:
        key = key.casefold()
    return key or None

gated_tagged_key(gate, value, *, tag, sep=':', prefix='')

:func:tagged_key behind a gate, for a routed source.

When one resource contributes several side-local keys — one per class its vertex_router collapses onto the aligned class — the router's discriminator selects which one applies. None when the gate does not match, which is an empty value to identity digests. This is the explicit, hand-written form (LocalKeySource.gate); a source keyed by member gets its gate derived from the router as a when guard on the step instead.

Parameters:

Name	Type	Description	Default
gate	str | None	Field deciding which branch this document is (the discriminator).	required
value	object	The side-local key material.	required
tag	str | None	Namespace prefix identifying the branch; empty for none.	required
sep	str	Separator between tag and the key.	':'
prefix	str	Required prefix of gate; "" always passes.	''

Returns:

Type	Description
str | None	f"{tag}{sep}{key}", or None when the gate fails or value is
str | None	missing or empty.

Source code in graflo/util/transform.py

def gated_tagged_key(
    gate: str | None,
    value: object,
    *,
    tag: str | None,
    sep: str = ":",
    prefix: str = "",
) -> str | None:
    """:func:`tagged_key` behind a gate, for a routed source.

    When one resource contributes several side-local keys — one per class its
    ``vertex_router`` collapses onto the aligned class — the router's
    discriminator selects which one applies. ``None`` when the gate does not
    match, which is an empty value to identity digests. This is the explicit,
    hand-written form (``LocalKeySource.gate``); a source keyed by member gets
    its gate derived from the router as a ``when`` guard on the step instead.

    Args:
        gate: Field deciding which branch this document is (the discriminator).
        value: The side-local key material.
        tag: Namespace prefix identifying the branch; empty for none.
        sep: Separator between *tag* and the key.
        prefix: Required prefix of *gate*; ``""`` always passes.

    Returns:
        ``f"{tag}{sep}{key}"``, or ``None`` when the gate fails or *value* is
        missing or empty.
    """
    if gate is None:
        return None
    if not str(gate).startswith(prefix):
        return None
    return tagged_key(value, tag=tag, sep=sep)

parse_date_conf(input_str)

Parse a date string in YYYYMMDD format.

Parameters:

Name	Type	Description	Default
input_str	str	Date string in YYYYMMDD format.	required

Returns:

Name	Type	Description
tuple	tuple[int, int, int]	(year, month, day) as integers.

Example

parse_date_conf("20230101") (2023, 1, 1)

Source code in graflo/util/transform.py

def parse_date_conf(input_str: str) -> tuple[int, int, int]:
    """Parse a date string in YYYYMMDD format.

    Args:
        input_str (str): Date string in YYYYMMDD format.

    Returns:
        tuple: (year, month, day) as integers.

    Example:
        >>> parse_date_conf("20230101")
        (2023, 1, 1)
    """
    dt = datetime.strptime(input_str, "%Y%m%d")
    return dt.year, dt.month, dt.day

parse_date_ibes(date0, time0)

Converts IBES date and time to ISO 8601 format datetime.

Parameters:

Name	Type	Description	Default
date0	str / int	Date in YYYYMMDD format.	required
time0	str	Time in HH:MM:SS format.	required

Returns:

Name	Type	Description
str	str	Datetime in ISO 8601 format (YYYY-MM-DDTHH:MM:SSZ).

Example

parse_date_ibes(20160126, "9:35:52") '2016-01-26T09:35:52Z'

Source code in graflo/util/transform.py

def parse_date_ibes(date0: str | int, time0: str) -> str:
    """Converts IBES date and time to ISO 8601 format datetime.

    Args:
        date0 (str/int): Date in YYYYMMDD format.
        time0 (str): Time in HH:MM:SS format.

    Returns:
        str: Datetime in ISO 8601 format (YYYY-MM-DDTHH:MM:SSZ).

    Example:
        >>> parse_date_ibes(20160126, "9:35:52")
        '2016-01-26T09:35:52Z'
    """
    date0 = str(date0)
    year, month, day = date0[:4], date0[4:6], date0[6:]
    full_datetime = f"{year}-{month}-{day}T{time0}Z"

    return full_datetime

parse_date_reference(input_str)

Extract year from a date reference string.

Parameters:

Name	Type	Description	Default
input_str	str	Date reference string.	required

Returns:

Name	Type	Description
int	int | str	Year from the date reference.

Example

parse_date_reference("1923, May 10") 1923

Source code in graflo/util/transform.py

def parse_date_reference(input_str: str) -> int | str:
    """Extract year from a date reference string.

    Args:
        input_str (str): Date reference string.

    Returns:
        int: Year from the date reference.

    Example:
        >>> parse_date_reference("1923, May 10")
        1923
    """
    return _parse_date_reference(input_str)["year"]

parse_date_standard(input_str)

Parse a date string in YYYY-MM-DD format.

Parameters:

Name	Type	Description	Default
input_str	str	Date string in YYYY-MM-DD format.	required

Returns:

Name	Type	Description
tuple	tuple[int, int, int]	(year, month, day) as integers.

Example

parse_date_standard("2023-01-01") (2023, 1, 1)

Source code in graflo/util/transform.py

def parse_date_standard(input_str: str) -> tuple[int, int, int]:
    """Parse a date string in YYYY-MM-DD format.

    Args:
        input_str (str): Date string in YYYY-MM-DD format.

    Returns:
        tuple: (year, month, day) as integers.

    Example:
        >>> parse_date_standard("2023-01-01")
        (2023, 1, 1)
    """
    dt = datetime.strptime(input_str, "%Y-%m-%d")
    return dt.year, dt.month, dt.day

parse_date_standard_to_epoch(input_str)

Convert standard date string to Unix epoch timestamp.

Parameters:

Name	Type	Description	Default
input_str	str	Date string in YYYY-MM-DD format.	required

Returns:

Name	Type	Description
float	float	Unix epoch timestamp.

Example

parse_date_standard_to_epoch("2023-01-01") 1672531200.0

Source code in graflo/util/transform.py

def parse_date_standard_to_epoch(input_str: str) -> float:
    """Convert standard date string to Unix epoch timestamp.

    Args:
        input_str (str): Date string in YYYY-MM-DD format.

    Returns:
        float: Unix epoch timestamp.

    Example:
        >>> parse_date_standard_to_epoch("2023-01-01")
        1672531200.0
    """
    dt = datetime.strptime(input_str, "%Y-%m-%d").timetuple()
    timestamp = time.mktime(dt)
    return timestamp

parse_date_yahoo(date0)

Convert Yahoo Finance date to ISO 8601 format.

Parameters:

Name	Type	Description	Default
date0	str	Date in YYYY-MM-DD format.	required

Returns:

Name	Type	Description
str	str	Datetime in ISO 8601 format with noon time.

Example

parse_date_yahoo("2023-01-01") '2023-01-01T12:00:00Z'

Source code in graflo/util/transform.py

def parse_date_yahoo(date0: str) -> str:
    """Convert Yahoo Finance date to ISO 8601 format.

    Args:
        date0 (str): Date in YYYY-MM-DD format.

    Returns:
        str: Datetime in ISO 8601 format with noon time.

    Example:
        >>> parse_date_yahoo("2023-01-01")
        '2023-01-01T12:00:00Z'
    """
    full_datetime = f"{date0}T12:00:00Z"
    return full_datetime

parse_multi_item(s, mapper, direct)

Parses complex multi-item strings into structured data.

Supports parsing strings with quoted or bracketed items.

Parameters:

Name	Type	Description	Default
s	str	Input string to parse.	required
mapper	dict	Mapping of input keys to output keys.	required
direct	list	Direct keys to extract.	required

Returns:

Name	Type	Description
defaultdict	defaultdict[str, list]	Parsed items with lists as values.

Example

Parsing '[name: John, age: 30] [name: Jane, age: 25]' with mapper={"name": "full_name"} and direct=["age"] produces grouped lists under each mapped key.

Source code in graflo/util/transform.py

def parse_multi_item(s: str, mapper: dict, direct: list) -> defaultdict[str, list]:
    """Parses complex multi-item strings into structured data.

    Supports parsing strings with quoted or bracketed items.

    Args:
        s (str): Input string to parse.
        mapper (dict): Mapping of input keys to output keys.
        direct (list): Direct keys to extract.

    Returns:
        defaultdict: Parsed items with lists as values.

    Example:
        Parsing ``'[name: John, age: 30] [name: Jane, age: 25]'`` with
        ``mapper={"name": "full_name"}`` and ``direct=["age"]`` produces grouped
        lists under each mapped key.
    """
    if "'" in s:
        items_str = re.findall(r"\"(.*?)\"", s) + re.findall(r"\'(.*?)\'", s)
    else:
        # remove brackets
        items_str = re.findall(r"\[([^]]+)", s)[0].split()
    r: defaultdict[str, list] = defaultdict(list)
    for item in items_str:
        doc0 = [ss.strip().split(":") for ss in item.split(",")]
        if all(len(x) == 2 for x in doc0):
            doc0_dict = dict(doc0)
            for n_init, n_final in mapper.items():
                try:
                    r[n_final] += [doc0_dict[n_init]]
                except KeyError:
                    r[n_final] += [None]

            for n_final in direct:
                # Use field.name for dictionary keys (JSON serialization requires strings)
                # Handle both Field objects and strings for backward compatibility
                key = n_final.name if hasattr(n_final, "name") else str(n_final)
                try:
                    r[key] += [doc0_dict[key]]
                except KeyError:
                    r[key] += [None]
        else:
            for key, value in zip(direct, doc0):
                # Use field.name for dictionary keys (JSON serialization requires strings)
                # Handle both Field objects and strings for backward compatibility
                key_str = key.name if hasattr(key, "name") else str(key)
                r[key_str] += [value]

    return r

pick_unique_dict(docs)

Remove duplicate structures from a list by content hash.

Uses a hash-based approach that handles nested dicts, lists/tuples, datetime objects, and Decimal types. Preserves original objects and insertion order. Works for vertex dicts and edge triples alike.

Parameters:

Name	Type	Description	Default
docs	list[T]	List of structures to deduplicate.	required

Returns:

Type	Description
list[T]	List of unique structures (preserving original objects).

Example

docs = [{"a": 1}, {"a": 1}, {"b": 2}] pick_unique_dict(docs) [{"a": 1}, {"b": 2}]

Source code in graflo/util/transform.py

def pick_unique_dict(docs: list[T]) -> list[T]:
    """Remove duplicate structures from a list by content hash.

    Uses a hash-based approach that handles nested dicts, lists/tuples,
    datetime objects, and Decimal types. Preserves original objects and
    insertion order. Works for vertex dicts and edge triples alike.

    Args:
        docs: List of structures to deduplicate.

    Returns:
        List of unique structures (preserving original objects).

    Example:
        >>> docs = [{"a": 1}, {"a": 1}, {"b": 2}]
        >>> pick_unique_dict(docs)
        [{"a": 1}, {"b": 2}]
    """
    from datetime import date, datetime, time
    from decimal import Decimal

    def make_hashable(obj):
        """Convert an object to a hashable representation.

        Handles nested structures, datetime objects, and Decimal types.

        Args:
            obj: Object to make hashable

        Returns:
            Hashable representation of the object
        """
        if isinstance(obj, dict):
            # Sort items by key for consistent hashing
            return tuple(sorted((k, make_hashable(v)) for k, v in obj.items()))
        elif isinstance(obj, (list, tuple)):
            return tuple(make_hashable(item) for item in obj)
        elif isinstance(obj, (datetime, date, time)):
            # Convert to ISO format string for hashing
            return ("__datetime__", obj.isoformat())
        elif isinstance(obj, Decimal):
            # Convert to string representation to preserve precision
            return ("__decimal__", str(obj))
        elif isinstance(obj, set):
            # Convert set to sorted tuple for consistent hashing
            return tuple(sorted(make_hashable(item) for item in obj))
        elif isinstance(obj, float) and math.isnan(obj):
            # NaN != NaN, so two NaN-carrying docs would never compare equal —
            # except by accident of object identity (the np.nan singleton),
            # which a pickle round-trip (worker processes) silently breaks.
            # Normalize to a marker so dedup follows value semantics.
            return ("__nan__",)
        else:
            # Primitive types (int, float, str, bool, None) are already hashable
            return obj

    # Use a dict to preserve insertion order and original objects
    seen: dict = {}
    for doc in docs:
        # Create hashable representation
        hashable_repr = make_hashable(doc)
        # Use hashable representation as key, original doc as value
        if hashable_repr not in seen:
            seen[hashable_repr] = doc

    # Return list of unique documents (preserving original objects)
    return list(seen.values())

remove_prefix(s, prefix) cached

Remove a prefix from a string key when present.

Source code in graflo/util/transform.py

@lru_cache(maxsize=32768)
def remove_prefix(s: str, prefix: str) -> str:
    """Remove a prefix from a string key when present."""
    return s.removeprefix(prefix)

remove_suffix(s, suffix) cached

Remove a suffix from a string key when present.

Source code in graflo/util/transform.py

@lru_cache(maxsize=32768)
def remove_suffix(s: str, suffix: str) -> str:
    """Remove a suffix from a string key when present."""
    return s.removesuffix(suffix)

round_str(x, **kwargs)

Round a string number to specified precision.

Parameters:

Name	Type	Description	Default
x	str	String representation of a number.	required
**kwargs		Additional arguments for round() function.	{}

Returns:

Name	Type	Description
float	float	Rounded number.

Example

round_str("3.14159", ndigits=2) 3.14

Source code in graflo/util/transform.py

def round_str(x: str, **kwargs) -> float:
    """Round a string number to specified precision.

    Args:
        x (str): String representation of a number.
        **kwargs: Additional arguments for round() function.

    Returns:
        float: Rounded number.

    Example:
        >>> round_str("3.14159", ndigits=2)
        3.14
    """
    return round(float(x), **kwargs)

snake_to_camel(s, upper_first=False) cached

Convert snake_case names to camelCase or PascalCase.

Source code in graflo/util/transform.py

@lru_cache(maxsize=32768)
def snake_to_camel(s: str, upper_first: bool = False) -> str:
    """Convert snake_case names to camelCase or PascalCase."""
    if not s:
        return s
    leading = _LEADING_UNDERSCORES_RE.match(s)
    trailing = _TRAILING_UNDERSCORES_RE.search(s)
    leading_part = leading.group(0) if leading else ""
    trailing_part = trailing.group(0) if trailing else ""

    core = s.strip("_")
    if not core:
        return s

    parts = [part for part in core.split("_") if part]
    if not parts:
        return s

    head = parts[0].capitalize() if upper_first else parts[0].lower()
    tail = "".join(part.capitalize() for part in parts[1:])
    return f"{leading_part}{head}{tail}{trailing_part}"

split_keep_part(s, sep='/', keep=-1)

Split a string and keep specified parts.

Parameters:

Name	Type	Description	Default
s	str	String to split.	required
sep	str	Separator to split on.	'/'
keep	int or list	Index or indices to keep.	-1

Returns:

Name	Type	Description
str	str	Joined string of kept parts.

Example

split_keep_part("a/b/c", keep=0) 'a' split_keep_part("a/b/c", keep=[0, 2]) 'a/c'

Source code in graflo/util/transform.py

def split_keep_part(s: str, sep="/", keep=-1) -> str:
    """Split a string and keep specified parts.

    Args:
        s (str): String to split.
        sep (str): Separator to split on.
        keep (int or list): Index or indices to keep.

    Returns:
        str: Joined string of kept parts.

    Example:
        >>> split_keep_part("a/b/c", keep=0)
        'a'
        >>> split_keep_part("a/b/c", keep=[0, 2])
        'a/c'
    """
    if isinstance(keep, list):
        items = s.split(sep)
        return sep.join(items[k] for k in keep)
    else:
        return s.split(sep)[keep]

standardize(k)

Standardizes a string key by removing periods and splitting.

Handles comma and space-separated strings, normalizing their format.

Parameters:

Name	Type	Description	Default
k	str	Input string to be standardized.	required

Returns:

Name	Type	Description
str	str	Cleaned and standardized string.

Example

standardize("John. Doe, Smith") 'John,Doe,Smith' standardize("John Doe Smith") 'John,Doe,Smith'

Source code in graflo/util/transform.py

def standardize(k: str) -> str:
    """Standardizes a string key by removing periods and splitting.

    Handles comma and space-separated strings, normalizing their format.

    Args:
        k (str): Input string to be standardized.

    Returns:
        str: Cleaned and standardized string.

    Example:
        >>> standardize("John. Doe, Smith")
        'John,Doe,Smith'
        >>> standardize("John Doe Smith")
        'John,Doe,Smith'
    """
    cleaned = k.translate(str.maketrans({".": ""}))
    # try to split by ", "
    parts = cleaned.split(", ")
    if len(parts) < 2:
        parts = parts[0].split(" ")
    else:
        parts[1] = parts[1].translate(str.maketrans({" ": ""}))
    return ",".join(parts)

tagged_key(value, *, tag, sep=':')

Namespace a side-local key: tag "a" turns "f2" into "a:f2".

None/empty value returns None, which is an empty value to identity digests — the funnel branch listing the output field is skipped. Disambiguation is resource knowledge: each resource tags its own local keys, so a class-level fallback identity stays side-agnostic while cross-resource collisions become impossible. Normalization beyond a whitespace strip is a separate concern — compose another transform step.

An empty tag ("" or None) is the neutral element: the key is returned as-is, with no separator. That is for values already unique across every source of the class — a UUID, an IRI, an id the source itself prefixes — where a namespace would only be noise.

Parameters:

Name	Type	Description	Default
value	object	The side-local key material.	required
tag	str | None	Namespace prefix identifying the resource/side; empty for none.	required
sep	str	Separator between tag and the key.	':'

Returns:

Type	Description
str | None	f"{tag}{sep}{key}" (or bare key under an empty tag), or
str | None	None when value is missing or empty.

Source code in graflo/util/transform.py

def tagged_key(value: object, *, tag: str | None, sep: str = ":") -> str | None:
    """Namespace a side-local key: tag ``"a"`` turns ``"f2"`` into ``"a:f2"``.

    ``None``/empty *value* returns ``None``, which is an empty value to
    identity digests — the funnel branch listing the output field is skipped.
    Disambiguation is resource knowledge: each resource tags its own local
    keys, so a class-level fallback identity stays side-agnostic while
    cross-resource collisions become impossible. Normalization beyond a
    whitespace strip is a separate concern — compose another transform step.

    An empty *tag* (``""`` or ``None``) is the neutral element: the key is
    returned as-is, with no separator. That is for values already unique
    across every source of the class — a UUID, an IRI, an id the source itself
    prefixes — where a namespace would only be noise.

    Args:
        value: The side-local key material.
        tag: Namespace prefix identifying the resource/side; empty for none.
        sep: Separator between *tag* and the key.

    Returns:
        ``f"{tag}{sep}{key}"`` (or bare ``key`` under an empty tag), or
        ``None`` when *value* is missing or empty.
    """
    if value is None:
        return None
    key = str(value).strip()
    if not key:
        return None
    return f"{tag}{sep}{key}" if tag else key

try_int(x)

Attempt to convert a value to integer.

Parameters:

Name	Type	Description	Default
x		Value to convert.	required

Returns:

Type	Description
	int or original value: Integer if conversion successful, original value otherwise.

Example

try_int("123") 123 try_int("abc") 'abc'

Source code in graflo/util/transform.py

def try_int(x):
    """Attempt to convert a value to integer.

    Args:
        x: Value to convert.

    Returns:
        int or original value: Integer if conversion successful, original value otherwise.

    Example:
        >>> try_int("123")
        123
        >>> try_int("abc")
        'abc'
    """
    try:
        x = int(x)
        return x
    except:
        return x