Utils

Snailz utilities.

UniqueIdGenerator

Generate unique IDs using provided function.

Source code in src/snailz/utils.py

class UniqueIdGenerator:
    """Generate unique IDs using provided function."""

    def __init__(self, name: str, func: Callable, limit: int = UNIQUE_ID_LIMIT) -> None:
        """Initialize.

        Parameters:
            name: A name for this generator
            func: Function that creates IDs when called
            limit: Maximum number of attempts
        """
        self._name = name
        self._func = func
        self._limit = limit
        self._seen = set()

    def next(self, *args: object) -> str:
        """Get next unique ID.

        Parameters:
            args: Arguments to pass to the ID-generating function

        Returns:
            A unique identifier that hasn't been returned before

        Raises:
            RuntimeError: If unable to generate a unique ID within limit attempts
        """
        for i in range(self._limit):
            ident = self._func(*args)
            if ident in self._seen:
                continue
            self._seen.add(ident)
            return ident
        raise RuntimeError(f"failed to find unique ID for {self._name}")

__init__(name, func, limit=UNIQUE_ID_LIMIT)

Initialize.

Parameters:

Name	Type	Description	Default
name	str	A name for this generator	required
func	Callable	Function that creates IDs when called	required
limit	int	Maximum number of attempts	UNIQUE_ID_LIMIT

Source code in src/snailz/utils.py

def __init__(self, name: str, func: Callable, limit: int = UNIQUE_ID_LIMIT) -> None:
    """Initialize.

    Parameters:
        name: A name for this generator
        func: Function that creates IDs when called
        limit: Maximum number of attempts
    """
    self._name = name
    self._func = func
    self._limit = limit
    self._seen = set()

next(*args)

Get next unique ID.

Parameters:

Name	Type	Description	Default
args	object	Arguments to pass to the ID-generating function	()

Returns:

Type	Description
str	A unique identifier that hasn't been returned before

Raises:

Type	Description
RuntimeError	If unable to generate a unique ID within limit attempts

Source code in src/snailz/utils.py

def next(self, *args: object) -> str:
    """Get next unique ID.

    Parameters:
        args: Arguments to pass to the ID-generating function

    Returns:
        A unique identifier that hasn't been returned before

    Raises:
        RuntimeError: If unable to generate a unique ID within limit attempts
    """
    for i in range(self._limit):
        ident = self._func(*args)
        if ident in self._seen:
            continue
        self._seen.add(ident)
        return ident
    raise RuntimeError(f"failed to find unique ID for {self._name}")

display(filepath, data)

Write to a file or to stdout.

Parameters:

Name	Type	Description	Default
filepath	str | None	Output filepath or None for stdout	required
data	BaseModel | str	what to write	required

Source code in src/snailz/utils.py

def display(filepath: str | None, data: BaseModel | str) -> None:
    """Write to a file or to stdout.

    Parameters:
        filepath: Output filepath or None for stdout
        data: what to write
    """
    if isinstance(data, str):
        text = data
    else:
        text = json.dumps(data, indent=2, default=_serialize_json)

    if not filepath:
        print(text)
    else:
        with open(filepath, "w") as writer:
            writer.write(text)

fail(msg)

Report failure and exit.

Parameters:

Name	Type	Description	Default
msg	str	Error message to display	required

Source code in src/snailz/utils.py

def fail(msg: str) -> None:
    """Report failure and exit.

    Parameters:
        msg: Error message to display
    """
    print(msg, file=sys.stderr)
    sys.exit(1)

report(verbose, msg)

Report if verbosity turned on.

Parameters:

Name	Type	Description	Default
verbose	bool	Is display on or off?	required
msg	str	Message to display	required

Source code in src/snailz/utils.py

def report(verbose: bool, msg: str) -> None:
    """Report if verbosity turned on.

    Parameters:
        verbose: Is display on or off?
        msg: Message to display
    """
    if verbose:
        print(msg)

to_csv(rows, fields, f_make_row)

Generic converter from list of models to CSV string.

Parameters:

Name	Type	Description	Default
rows	list	List of rows to convert.	required
fields	list	List of names of columns.	required
f_make_row	Callable	Function that converts a row to text.	required

Returns:

Type	Description
str	CSV representation of data.

Source code in src/snailz/utils.py

def to_csv(rows: list, fields: list, f_make_row: Callable) -> str:
    """Generic converter from list of models to CSV string.

    Parameters:
        rows: List of rows to convert.
        fields: List of names of columns.
        f_make_row: Function that converts a row to text.

    Returns:
        CSV representation of data.
    """

    output = io.StringIO()
    writer = csv.writer(output, lineterminator="\n")
    writer.writerow(fields)
    for r in rows:
        writer.writerow(f_make_row(r))
    return output.getvalue()

_serialize_json(obj)

Custom JSON serializer for JSON conversion.

Parameters:

Name	Type	Description	Default
obj	object	The object to serialize	required

Returns:

Type	Description
str | dict	String representation of date objects or dict for Pydantic models

Raises:

Type	Description
TypeError	If the object type is not supported for serialization

Source code in src/snailz/utils.py

def _serialize_json(obj: object) -> str | dict:
    """Custom JSON serializer for JSON conversion.

    Parameters:
        obj: The object to serialize

    Returns:
        String representation of date objects or dict for Pydantic models

    Raises:
        TypeError: If the object type is not supported for serialization
    """
    if isinstance(obj, date):
        return obj.isoformat()
    if isinstance(obj, BaseModel):
        return obj.model_dump()
    raise TypeError(f"Type {type(obj)} not serializable")