Skip to content
GitHubDiscord

Utilities

Helper functions and utilities for normalization, value providers, and parameter injection.


Utilities for normalizing strings and data structures using Unicode normalization forms.

Module: giskard.checks.utils.normalization

Normalize a string using a specified Unicode normalization form.

from giskard.checks.utils.normalization import normalize_string
# NFC normalization (canonical composition)
text = "café" # with combining accent
normalized = normalize_string(text, normalization_form="NFC")
# NFD normalization (canonical decomposition)
normalized = normalize_string(text, normalization_form="NFD")
# NFKC normalization (compatibility composition)
normalized = normalize_string(text, normalization_form="NFKC")
# NFKD normalization (compatibility decomposition)
normalized = normalize_string(text, normalization_form="NFKD")

Parameters:

ParameterTypeDefaultDescription
valuestrrequiredString to normalize
normalization_formLiteral['NFC', 'NFD', 'NFKC', 'NFKD'] | NoneNoneUnicode normalization form

Returns:

  • str: Normalized string

Recursively normalize all strings in a dictionary or list.

from giskard.checks.utils.normalization import normalize_data
# Normalize nested data structure
data = {
"name": "café",
"items": ["naïve", "résumé"],
"details": {
"city": "São Paulo"
}
}
normalized_data = normalize_data(data, normalization_form="NFC")

Parameters:

ParameterTypeDefaultDescription
dataT (dict or list)requiredData structure to normalize
normalization_formLiteral['NFC', 'NFD', 'NFKC', 'NFKD'] | NoneNoneUnicode normalization form

Returns:

  • T: Data structure with normalized strings
FormDescriptionUse Case
NFCCanonical CompositionMost common; combines characters
NFDCanonical DecompositionSeparates base + combining characters
NFKCCompatibility CompositionCombines + converts compatibility chars
NFKDCompatibility DecompositionSeparates + converts compatibility chars

Abstract value provider patterns for static values and callables.

Module: giskard.checks.utils.value_provider

Abstract base class for value providers.

from giskard.checks.utils.value_provider import ValueProvider
class CustomProvider(ValueProvider):
"""Custom value provider implementation."""
async def provide(self, *args, **kwargs):
# Custom logic to provide value
return computed_value

Provider for static values.

from giskard.checks.utils.value_provider import StaticValueProvider
# Create a static value provider
provider = StaticValueProvider(value="hello")
# Provide the value
result = await provider.provide() # Returns "hello"

Parameters:

ParameterTypeDescription
valueRThe static value to provide

Provider that wraps a callable with parameter injection.

from giskard.checks.utils.value_provider import CallableValueProvider
from giskard.checks.utils.parameter_injection import CallableInjectionMapping
def compute_value(x: int, y: int) -> int:
return x + y
# Create injection mapping
mapping = CallableInjectionMapping.from_callable(
compute_value,
# Define parameter injections here
)
# Create provider
provider = CallableValueProvider(
callable=compute_value,
injection_mapping=mapping
)
# Provide the value
result = await provider.provide()

Parameters:

ParameterTypeDescription
callableCallableFunction to call for value
injection_mappingCallableInjectionMappingParameter injection configuration

Provider for generator-based values.

from giskard.checks.utils.value_provider import ValueGeneratorProvider
async def value_generator():
yield "first"
yield "second"
yield "third"
provider = ValueGeneratorProvider.from_mapping(value_generator)
# Use the provider
async for value in provider.provide():
print(value)

Dynamic parameter injection for callables.

Module: giskard.checks.utils.parameter_injection

Maps parameters to inject into a callable.

from giskard.checks.utils.parameter_injection import CallableInjectionMapping
def my_function(trace, inputs, config):
# Function that needs parameters injected
return process(trace, inputs, config)
# Create injection mapping
mapping = CallableInjectionMapping.from_callable(
my_function,
# Define which parameters to inject and from where
)
# Inject parameters
bound_function = mapping.inject_parameters(
my_function,
trace=current_trace,
inputs=current_inputs,
config=current_config
)
# Call with injected parameters
result = bound_function()

Methods:

from_callable(callable, *args_reqs, **kwargs_reqs)

Section titled “from_callable(callable, *args_reqs, **kwargs_reqs)”

Create an injection mapping from a callable.

Parameters:

  • callable: The function to create mapping for
  • args_reqs: Positional argument injection requirements
  • kwargs_reqs: Keyword argument injection requirements

Returns: CallableInjectionMapping

Inject parameters into a callable.

Parameters:

  • value: Callable to inject into
  • args: Positional arguments to inject
  • kwargs: Keyword arguments to inject

Returns: Bound callable with injected parameters

Configuration for a single parameter injection.

from giskard.checks.utils.parameter_injection import ParameterInjection
# Create parameter injection configuration
injection = ParameterInjection(
name="trace",
source="context"
)

Utilities for converting values and generators to async generators.

Module: giskard.checks.utils.generator

Convert a value or generator (sync or async) into an async generator.

from giskard.checks.utils.generator import a_generator
# Convert static value to async generator
async for value in a_generator("hello"):
print(value) # Yields "hello" once
# Convert sync generator
def sync_gen():
yield 1
yield 2
yield 3
async for value in a_generator(sync_gen()):
print(value) # Yields 1, 2, 3
# Convert async generator (pass-through)
async def async_gen():
yield "a"
yield "b"
async for value in a_generator(async_gen()):
print(value) # Yields "a", "b"

Parameters:

ParameterTypeDescription
value_or_generatorYieldType | SyncOrAsyncGeneratorValue or generator to convert

Returns:

  • AsyncGenerator: Async generator yielding values

Use Cases:

# Normalize different input types to async generators
async def process_values(source):
async for value in a_generator(source):
# Process each value
result = transform(value)
yield result
# Works with static values
async for r in process_values("single value"):
print(r)
# Works with lists
async for r in process_values([1, 2, 3]):
print(r)
# Works with generators
async for r in process_values(my_generator()):
print(r)

from giskard.checks import Check, CheckResult, Trace
from giskard.checks.utils.normalization import normalize_string
@Check.register("normalized_comparison")
class NormalizedComparison(Check):
expected: str
normalization_form: str = "NFC"
async def run(self, trace: Trace) -> CheckResult:
actual = trace.last.outputs
# Normalize both strings
expected_norm = normalize_string(
self.expected,
normalization_form=self.normalization_form
)
actual_norm = normalize_string(
actual,
normalization_form=self.normalization_form
)
if expected_norm == actual_norm:
return CheckResult.success(
message="Strings match after normalization"
)
else:
return CheckResult.failure(
message="Strings don't match",
details={
"expected": expected_norm,
"actual": actual_norm
}
)
from giskard.checks.utils.value_provider import ValueProvider
class DatabaseValueProvider(ValueProvider):
"""Provide values from a database."""
def __init__(self, query: str, db_connection):
self.query = query
self.db = db_connection
async def provide(self, **params):
# Execute query with parameters
result = await self.db.execute(self.query, **params)
return result.fetchone()
# Use in testing
provider = DatabaseValueProvider(
query="SELECT name FROM users WHERE id = :user_id",
db_connection=db
)
# Get value
user_name = await provider.provide(user_id=123)