uf #

Returns:

True if valid

class uf.AsyncFunctionWrapper(func)[source]#

Bases: object

Wrapper for async functions that provides both sync and async access.

This allows the same function to be called either way, depending on the context.

Example

>>> async def fetch_user(user_id: int):
...     # async database query
...     return user_data
>>>
>>> wrapper = AsyncFunctionWrapper(fetch_user)
>>> # Sync call
>>> user = wrapper.call_sync(user_id=123)
>>> # Async call
>>> user = await wrapper.call_async(user_id=123)

async call_async(*args, **kwargs)[source]#

Call the function asynchronously.

Parameters:

*args – Positional arguments
**kwargs – Keyword arguments

Return type:

Returns:

Function result

call_sync(*args, **kwargs)[source]#

Call the function synchronously.

Parameters:

*args – Positional arguments
**kwargs – Keyword arguments

Return type:

Returns:

Function result

class uf.AuthBackend[source]#

Bases: object

Base authentication backend.

Subclass this to create custom authentication backends.

authenticate(username, password)[source]#

Authenticate a user.

Parameters:

username (str) – Username
password (str) – Password

Return type:

Optional[User]

Returns:

User object if authenticated, None otherwise

create_user(username, password, roles=None, permissions=None, **metadata)[source]#

Create a new user.

Parameters:

username (str) – Username
password (str) – Plain text password
roles (Optional[list[str]]) – List of roles
permissions (Optional[list[str]]) – List of permissions
**metadata – Additional user metadata

Return type:

User

Returns:

Created User object

delete_user(username)[source]#

Delete a user.

Parameters:: username (str) – Username
Return type:: bool
Returns:: True if deleted successfully

get_user(username)[source]#

Get a user by username.

Parameters:: username (str) – Username
Return type:: Optional[User]
Returns:: User object or None

update_user(username, **updates)[source]#

Update user information.

Parameters:

username (str) – Username
**updates – Fields to update

Return type:

Returns:

True if updated successfully

class uf.CacheBackend[source]#

Bases: object

Base class for cache backends.

clear()[source]#

Clear all cache entries.

Return type:: None

delete(key)[source]#

Delete a key from cache.

Parameters:: key (str) – Cache key
Return type:: bool
Returns:: True if deleted

exists(key)[source]#

Check if key exists in cache.

Parameters:: key (str) – Cache key
Return type:: bool
Returns:: True if exists and not expired

get(key)[source]#

Get a value from cache.

Parameters:: key (str) – Cache key
Return type:: Optional[Any]
Returns:: Cached value or None if not found/expired

set(key, value, ttl=None)[source]#

Set a value in cache.

Parameters:

key (str) – Cache key
value (Any) – Value to cache
ttl (Optional[int]) – Time to live in seconds

Return type:

class uf.CallHistory(max_size=100)[source]#

Bases: object

Manage history of function calls.

Example

>>> history = CallHistory(max_size=100)
>>> history.record('add', {'x': 10, 'y': 20}, result=30)
>>> recent = history.get_recent('add', limit=5)

clear(func_name=None)[source]#

Clear history.

Parameters:: func_name (Optional[str]) – Function to clear, or None for all
Return type:: None

classmethod from_dict(data, max_size=100)[source]#

Create from dictionary.

Parameters:

data (dict) – Dictionary with history data
max_size (int) – Maximum size per function

Return type:

CallHistory

Returns:

CallHistory instance

get_recent(func_name, limit=10)[source]#

Get recent calls for a function.

Parameters:

func_name (str) – Function name
limit (int) – Maximum number to return

Return type:

list[FunctionCall]

Returns:

List of recent FunctionCall objects (newest first)

get_successful_calls(func_name, limit=10)[source]#

Get recent successful calls.

Parameters:

func_name (str) – Function name
limit (int) – Maximum number to return

Return type:

list[FunctionCall]

Returns:

List of successful FunctionCall objects

record(func_name, params, result=None, success=True, error=None)[source]#

Record a function call.

Parameters:

func_name (str) – Name of the function
params (dict) – Parameters used
result (Any) – Result returned
success (bool) – Whether call succeeded
error (Optional[str]) – Error message if failed

Return type:

to_dict()[source]#

Convert history to dictionary.

Return type:: dict
Returns:: Dictionary representation

class uf.ConditionalFieldConfig(field_name, *, condition, then_schema=None, else_schema=None)[source]#

Bases: object

Configuration for conditional field display.

Allows fields to be shown/hidden based on the values of other fields.

Example

>>> # Show 'other_reason' field only when reason is 'other'
>>> config = ConditionalFieldConfig(
...     'other_reason',
...     condition={'reason': {'const': 'other'}}
... )

to_json_schema()[source]#

Convert to JSON Schema if/then/else structure.

Return type:: dict
Returns:: JSON Schema conditional structure

class uf.DependencyAction(*values)[source]#

Bases: Enum

Actions that can be triggered by field dependencies.

class uf.DependencyBuilder[source]#

Bases: object

Builder for creating field dependencies.

Provides a fluent interface for defining dependencies between fields.

Example

>>> builder = DependencyBuilder()
>>> builder.when('reason').equals('other').show('other_reason')
>>> builder.when('age').greater_than(18).enable('alcohol_consent')
>>> dependencies = builder.build()

build()[source]#

Build and return the list of dependencies.

Return type:: list[FieldDependency]
Returns:: List of FieldDependency objects

custom(condition)[source]#

Condition: custom callable.

Parameters:: condition (Callable[[Any], bool]) – Function that takes value and returns bool
Return type:: DependencyBuilder
Returns:: Self for method chaining

disable(target_field)[source]#

Action: disable target field.

Parameters:: target_field (str) – Name of field to disable
Return type:: DependencyBuilder
Returns:: Self for method chaining

enable(target_field)[source]#

Action: enable target field.

Parameters:: target_field (str) – Name of field to enable
Return type:: DependencyBuilder
Returns:: Self for method chaining

equals(value)[source]#

Condition: field equals value.

Parameters:: value (Any) – Value to compare against
Return type:: DependencyBuilder
Returns:: Self for method chaining

greater_than(value)[source]#

Condition: field is greater than value.

Parameters:: value (Any) – Value to compare against
Return type:: DependencyBuilder
Returns:: Self for method chaining

hide(target_field)[source]#

Action: hide target field.

Parameters:: target_field (str) – Name of field to hide
Return type:: DependencyBuilder
Returns:: Self for method chaining

in_list(values)[source]#

Condition: field value is in list.

Parameters:: values (list) – List of acceptable values
Return type:: DependencyBuilder
Returns:: Self for method chaining

is_falsy()[source]#

Condition: field is falsy.

Return type:: DependencyBuilder
Returns:: Self for method chaining

is_truthy()[source]#

Condition: field is truthy.

Return type:: DependencyBuilder
Returns:: Self for method chaining

less_than(value)[source]#

Condition: field is less than value.

Parameters:: value (Any) – Value to compare against
Return type:: DependencyBuilder
Returns:: Self for method chaining

make_optional(target_field)[source]#

Action: make target field optional.

Parameters:: target_field (str) – Name of field to make optional
Return type:: DependencyBuilder
Returns:: Self for method chaining

not_equals(value)[source]#

Condition: field does not equal value.

Parameters:: value (Any) – Value to compare against
Return type:: DependencyBuilder
Returns:: Self for method chaining

require(target_field)[source]#

Action: make target field required.

Parameters:: target_field (str) – Name of field to require
Return type:: DependencyBuilder
Returns:: Self for method chaining

show(target_field)[source]#

Action: show target field.

Parameters:: target_field (str) – Name of field to show
Return type:: DependencyBuilder
Returns:: Self for method chaining

when(field_name)[source]#

Start a dependency condition on a field.

Parameters:: field_name (str) – Name of the source field
Return type:: DependencyBuilder
Returns:: Self for method chaining

class uf.DictAuthBackend[source]#

Bases: AuthBackend

Simple in-memory dictionary-based authentication.

Suitable for development and simple applications.

Example

>>> backend = DictAuthBackend()
>>> backend.create_user('admin', 'secret', roles=['admin'])
>>> user = backend.authenticate('admin', 'secret')

authenticate(username, password)[source]#

Authenticate a user.

Return type:: Optional[User]

create_user(username, password, roles=None, permissions=None, **metadata)[source]#

Create a new user.

Return type:: User

delete_user(username)[source]#

Delete a user.

Return type:: bool

classmethod from_dict(users_data)[source]#

Create backend from dictionary.

Parameters:: users_data (dict) – Dictionary mapping usernames to user info
Return type:: DictAuthBackend

Example

>>> backend = DictAuthBackend.from_dict({
...     'admin': {'password': 'secret', 'roles': ['admin']},
...     'user': {'password': 'pass', 'roles': ['user']},
... })

get_user(username)[source]#

Get a user by username.

Return type:: Optional[User]

update_user(username, **updates)[source]#

Update user information.

Return type:: bool

class uf.DiskCache(cache_dir='.uf_cache', default_ttl=3600)[source]#

Bases: CacheBackend

Disk-based cache backend using pickle.

Persists cache to disk, suitable for larger datasets or persistence across restarts.

Example

>>> cache = DiskCache(cache_dir='/tmp/uf_cache')
>>> cache.set('expensive_result', big_data)

clear()[source]#

Clear all cache entries.

Return type:: None

delete(key)[source]#

Delete a key from cache.

Return type:: bool

get(key)[source]#

Get a value from cache.

Return type:: Optional[Any]

set(key, value, ttl=None)[source]#

Set a value in cache.

Return type:: None

class uf.FieldDependency(source_field, target_field, condition, action, else_action=None)[source]#

Bases: object

Define a dependency between form fields.

source_field#: Name of the field that triggers the dependency

target_field#: Name of the field affected by the dependency

condition#: Function that takes source value and returns bool

action#: Action to perform when condition is true

else_action#: Optional action when condition is false

Example

>>> # Show 'other_reason' only when reason is 'other'
>>> dep = FieldDependency(
...     source_field='reason',
...     target_field='other_reason',
...     condition=lambda v: v == 'other',
...     action=DependencyAction.SHOW,
...     else_action=DependencyAction.HIDE
... )

check(value)[source]#

Check the condition and return the appropriate action.

Parameters:: value (Any) – Value of the source field
Return type:: DependencyAction
Returns:: The action to perform

to_json_schema()[source]#

Convert to JSON Schema dependencies format.

Return type:: dict
Returns:: JSON Schema dependencies structure

Note

This uses JSON Schema’s if/then/else structure for dependencies.

to_ui_schema()[source]#

Convert to RJSF UI Schema dependencies format.

Return type:: dict
Returns:: UI Schema dependencies structure

class uf.FormDataBuilder[source]#

Bases: object

Builder for constructing form data for tests.

Provides a fluent interface for building test form data.

Example

>>> form_data = (
...     FormDataBuilder()
...     .field('name', 'John Doe')
...     .field('email', 'john@example.com')
...     .field('age', 30)
...     .build()
... )

build()[source]#

Build and return the form data.

Return type:: dict
Returns:: Dictionary of form data

field(name, value)[source]#

Add a field to the form data.

Parameters:

name (str) – Field name
value (Any) – Field value

Return type:

FormDataBuilder

Returns:

Self for method chaining

fields(**kwargs)[source]#

Add multiple fields at once.

Parameters:: **kwargs – Field name-value pairs
Return type:: FormDataBuilder
Returns:: Self for method chaining

class uf.FunctionCall(func_name, params, result=None, success=True, error=None)[source]#

Bases: object

Record of a single function call.

func_name#: Name of the function

params#: Parameters used

result#: Result returned (if captured)

timestamp#: When the call was made

success#: Whether the call succeeded

error#: Error message if failed

classmethod from_dict(data)[source]#

Create from dictionary.

Parameters:: data (dict) – Dictionary with call data
Return type:: FunctionCall
Returns:: FunctionCall instance

to_dict()[source]#

Convert to dictionary.

Return type:: dict
Returns:: Dictionary representation

class uf.FunctionGroup(name, funcs=<factory>, description='', icon=None, order=0, collapsed=False)[source]#

Bases: object

Group of functions with metadata.

name#: Name of the group

funcs#: Functions in this group: callables (partials and callable instances are fine; they are named as uf.specs.name_of_func does), or a {name: callable} mapping to name them explicitly

description#: Description of the group

icon#: Optional icon identifier for the group

order#: Display order (lower numbers first)

collapsed#: Whether the group starts collapsed in UI

Example

>>> math_funcs = FunctionGroup(
...     'Math',
...     [add, subtract, multiply, divide],
...     description='Mathematical operations',
...     icon='calculator'
... )

add_function(func, *, name=None)[source]#

Add a function to this group.

Parameters:

func (Callable) – Function to add
name (Optional[str]) – Explicit name to serve it under (default: its own name). Giving one turns funcs into a {name: callable} mapping.

Return type:

FunctionGroup

Returns:

Self for method chaining

Raises:

ValueError – If a different callable already has that name in the group

get_function_names()[source]#

Get names of all functions in this group.

Return type:: list[str]
Returns:: List of function name strings

property named_funcs: dict[str, Callable]#

{name: callable} of this group, named the way the app serves them.

>>> import functools
>>> g = FunctionGroup('G', [len, functools.partial(int, base=2)])
>>> list(g.named_funcs)
['len', 'int']

class uf.FunctionOrganizer[source]#

Bases: object

Organize functions into groups and hierarchies.

This class provides a fluent interface for building function organization structures that can be used to generate grouped navigation in the UI.

Example

>>> organizer = FunctionOrganizer()
>>> organizer.group('Admin', [user_create, user_delete], icon='shield')
>>> organizer.group('Reports', [generate_report, export_csv], icon='file')
>>> groups = organizer.get_groups()

add_to_group(group_name, func)[source]#

Add a function to an existing group.

Parameters:

group_name (str) – Name of the group
func (Callable) – Function to add

Return type:

FunctionOrganizer

Returns:

Self for method chaining

Raises:

ValueError – If group doesn’t exist

add_ungrouped(func)[source]#

Add a function without a group.

Parameters:: func (Callable) – Function to add
Return type:: FunctionOrganizer
Returns:: Self for method chaining

get_all_functions()[source]#

Get all functions across all groups.

Return type:: list[Callable]
Returns:: List of all functions

get_groups()[source]#

Get all groups, sorted by order.

Return type:: list[FunctionGroup]
Returns:: List of FunctionGroup objects sorted by order

group(name, funcs=None, *, description='', icon=None, order=0, collapsed=False)[source]#

Create and add a function group.

Parameters:

name (str) – Name of the group
funcs (Optional[Iterable[Callable]]) – Optional functions to add to group
description (str) – Description of the group
icon (Optional[str]) – Optional icon identifier
order (int) – Display order
collapsed (bool) – Whether to start collapsed

Return type:

FunctionGroup

Returns:

The created FunctionGroup

Example

>>> organizer.group(
...     'Database',
...     [save_record, load_record],
...     description='Database operations',
...     icon='database'
... )

to_dict()[source]#

Convert organization to dictionary format.

Return type:: dict
Returns:: Dictionary representation suitable for JSON serialization

Example

>>> org_dict = organizer.to_dict()
>>> # Can be used in templates or APIs

class uf.FunctionSpecStore(funcs, *, rjsf_config=None, ui_schema_factory=None, param_to_prop_type=None, validate=False)[source]#

Bases: Mapping

A mapping from function names to their RJSF specifications.

Lazily generates and caches form specs for each function.

Parameters:

funcs (Union[Iterable[Callable], Mapping[str, Callable]]) – Iterable of callables to generate specs for (named by name_of_func: __name__, the wrapped function’s for a functools.partial, the class name for a callable instance), or a mapping {name: callable} to name them explicitly
rjsf_config (Optional[dict]) – Optional configuration dict for RJSF generation
ui_schema_factory (Optional[Callable]) – Optional callable to customize UI schema generation
param_to_prop_type (Optional[Callable]) – Optional callable to map parameters to property types
validate (bool) – If true, every function is checked and its spec generated at construction, so a callable uf can’t build a form for fails there, naming it, rather than on the first request (default false: checked on first access of its spec; mk_rjsf_app validates).

Example

>>> def add(x: int, y: int) -> int:
...     '''Add two numbers'''
...     return x + y
>>> specs = FunctionSpecStore([add])
>>> 'add' in specs
True
>>> spec = specs['add']
>>> 'schema' in spec
True

property function_list: list[dict]#

Get list of all functions with basic metadata.

Returns:: List of dictionaries with function name and description

get_func(func_name)[source]#

Get the original function object by name.

Parameters:: func_name (str) – Name of the function
Return type:: Callable
Returns:: The function object
Raises:: KeyError – If function name not found

class uf.HistoryManager(max_history=100)[source]#

Bases: object

Combined manager for history and presets.

Provides a unified interface for tracking calls and managing presets.

Example

>>> manager = HistoryManager()
>>> manager.record_call('add', {'x': 10, 'y': 20}, result=30)
>>> manager.save_preset('quick', 'add', {'x': 10, 'y': 20})
>>> recent = manager.get_recent_calls('add')
>>> presets = manager.get_presets('add')

get_preset(preset_name, func_name)[source]#

Get a preset.

Parameters:

preset_name (str) – Name of preset
func_name (str) – Function name

Return type:

Optional[Preset]

Returns:

Preset object or None

get_presets(func_name)[source]#

Get all presets for a function.

Parameters:: func_name (str) – Function name
Return type:: list[Preset]
Returns:: List of Preset objects

get_recent_calls(func_name, limit=10)[source]#

Get recent calls.

Parameters:

func_name (str) – Function name
limit (int) – Maximum number to return

Return type:

list[FunctionCall]

Returns:

List of recent FunctionCall objects

classmethod load_from_file(filepath, max_history=100)[source]#

Load from file.

Parameters:

filepath (str) – Path to load from
max_history (int) – Maximum history size

Return type:

HistoryManager

Returns:

HistoryManager instance

record_call(func_name, params, result=None, success=True, error=None)[source]#

Record a function call.

Parameters:

func_name (str) – Function name
params (dict) – Parameters used
result (Any) – Result returned
success (bool) – Whether call succeeded
error (Optional[str]) – Error message if failed

Return type:

save_preset(preset_name, func_name, params, description='')[source]#

Save a preset.

Parameters:

preset_name (str) – Name for preset
func_name (str) – Function name
params (dict) – Parameter values
description (str) – Optional description

Return type:

Preset

Returns:

Created Preset object

save_to_file(filepath)[source]#

Save history and presets to file.

Parameters:: filepath (str) – Path to save to
Return type:: None

class uf.InputTransformRegistry[source]#

Bases: object

Registry for custom type transformations.

Integrates with qh’s type registry and extends it for UI needs, allowing custom types to be properly handled in both the form interface and the HTTP service layer.

Example

>>> from datetime import datetime
>>> registry = InputTransformRegistry()
>>>
>>> # Register a custom type
>>> registry.register_type(
...     datetime,
...     to_json=lambda dt: dt.isoformat(),
...     from_json=lambda s: datetime.fromisoformat(s),
...     ui_widget='datetime'
... )

from_json(value, py_type)[source]#

Transform a JSON value to Python type.

Parameters:

value (Any) – JSON value to transform
py_type (Type) – Target Python type

Return type:

Returns:

Transformed value

get_all_registered_types()[source]#

Get list of all registered types.

Return type:: list[Type]
Returns:: List of registered Python types

get_handler(py_type)[source]#

Get handler for a type.

Parameters:: py_type (Type) – Python type to look up
Return type:: Optional[dict]
Returns:: Handler dict or None if not registered

get_ui_config(py_type)[source]#

Get UI configuration for a type.

Parameters:: py_type (Type) – Python type to look up
Return type:: Optional[RjsfFieldConfig]
Returns:: RjsfFieldConfig or None if not registered

mk_input_trans_for_funcs(funcs)[source]#

Create input transformation compatible with qh.

This creates a transformation function that can be passed to qh.mk_app as the input_trans parameter.

Parameters:: funcs (list[Callable]) – List of functions to create transformation for
Return type:: Callable
Returns:: Transformation function for qh

Example

>>> from uf import mk_rjsf_app
>>> registry = InputTransformRegistry()
>>> # ... register types ...
>>> input_trans = registry.mk_input_trans_for_funcs([my_func])
>>> app = mk_rjsf_app([my_func], input_trans=input_trans)

mk_output_trans()[source]#

Create output transformation for qh.

Return type:: Callable
Returns:: Transformation function for qh output

Example

>>> output_trans = registry.mk_output_trans()
>>> app = mk_rjsf_app([my_func], output_trans=output_trans)

register_type(py_type, *, to_json=None, from_json=None, ui_widget=None, ui_config=None, json_schema_type=None, json_schema_format=None)[source]#

Parameters:

py_type (Type) – Python type to register
to_json (Optional[Callable[[Any], Any]]) – Function to convert Python type to JSON-serializable
from_json (Optional[Callable[[Any], Any]]) – Function to convert JSON to Python type
ui_widget (Optional[str]) – RJSF widget to use for this type
ui_config (Optional[RjsfFieldConfig]) – Full RjsfFieldConfig for this type
json_schema_type (Optional[str]) – JSON Schema type (e.g., ‘string’, ‘number’)
json_schema_format (Optional[str]) – JSON Schema format (e.g., ‘date-time’, ‘email’)

Return type:

Example

>>> from pathlib import Path
>>> registry.register_type(
...     Path,
...     to_json=str,
...     from_json=Path,
...     ui_widget='text',
...     json_schema_type='string'
... )

to_json(value, py_type=None)[source]#

Transform a Python value to JSON-serializable form.

Parameters:

value (Any) – Value to transform
py_type (Optional[Type]) – Optional type hint (uses type(value) if not provided)

Return type:

Returns:

JSON-serializable value

class uf.MemoryCache(default_ttl=3600, max_size=1000)[source]#

Bases: CacheBackend

In-memory cache backend.

Simple dictionary-based caching suitable for single-process applications.

Example

>>> cache = MemoryCache(default_ttl=3600)
>>> cache.set('key', 'value', ttl=60)
>>> value = cache.get('key')

cleanup_expired()[source]#

Remove expired entries.

Return type:: int
Returns:: Number of entries removed

clear()[source]#

Clear all cache entries.

Return type:: None

delete(key)[source]#

Delete a key from cache.

Return type:: bool

get(key)[source]#

Get a value from cache.

Return type:: Optional[Any]

set(key, value, ttl=None)[source]#

Set a value in cache.

Return type:: None

stats()[source]#

Get cache statistics.

Return type:: dict
Returns:: Dictionary with cache stats

class uf.OpenAPIConfig(title='API', version='1.0.0', description='', servers=None, enable_swagger=True, enable_redoc=True)[source]#

Bases: object

Configuration for OpenAPI generation.

Example

>>> config = OpenAPIConfig(
...     title="My API",
...     version="2.0.0",
...     description="API for my application"
... )

to_dict()[source]#

Convert to dictionary.

Return type:: dict
Returns:: Configuration dictionary

class uf.Preset(name, func_name, params, description='')[source]#

Bases: object

A saved parameter preset for a function.

name#: Preset name

func_name#: Function this preset is for

params#: Parameter values

description#: Optional description

created_at#: When preset was created

classmethod from_dict(data)[source]#

Create from dictionary.

Parameters:: data (dict) – Dictionary with preset data
Return type:: Preset
Returns:: Preset instance

to_dict()[source]#

Convert to dictionary.

Return type:: dict
Returns:: Dictionary representation

class uf.PresetManager[source]#

Bases: object

Manage parameter presets for functions.

Example

>>> presets = PresetManager()
>>> presets.save('quick_add', 'add', {'x': 10, 'y': 20}, 'Quick test')
>>> preset = presets.get('quick_add', 'add')
>>> result = my_func(**preset.params)

delete(preset_name, func_name)[source]#

Delete a preset.

Parameters:

preset_name (str) – Name of preset
func_name (str) – Function name

Return type:

Returns:

True if deleted, False if not found

classmethod from_dict(data)[source]#

Create from dictionary.

Parameters:: data (dict) – Dictionary with preset data
Return type:: PresetManager
Returns:: PresetManager instance

get(preset_name, func_name)[source]#

Get a preset.

Parameters:

preset_name (str) – Name of preset
func_name (str) – Function name

Return type:

Optional[Preset]

Returns:

Preset object or None

list_presets(func_name)[source]#

List all presets for a function.

Parameters:: func_name (str) – Function name
Return type:: list[Preset]
Returns:: List of Preset objects

save(preset_name, func_name, params, description='')[source]#

Save a parameter preset.

Parameters:

preset_name (str) – Name for this preset
func_name (str) – Function this preset is for
params (dict) – Parameter values
description (str) – Optional description

Return type:

Preset

Returns:

Created Preset object

to_dict()[source]#

Convert to dictionary.

Return type:: dict
Returns:: Dictionary representation

class uf.ResultRenderer[source]#

Bases: object

Base class for result renderers.

Subclass this to create custom renderers for specific result types.

can_render(result)[source]#

Check if this renderer can handle the given result.

Parameters:: result (Any) – The function result to render
Return type:: bool
Returns:: True if this renderer can handle the result

render(result)[source]#

Render the result to a displayable format.

Parameters:

result (Any) – The function result to render

Returns:

‘type’: Renderer type (e.g., ‘table’, ‘chart’, ‘json’)
’data’: Rendered data
’options’: Optional rendering options

Return type:

class uf.ResultRendererRegistry[source]#

Bases: object

Registry for result renderers.

Manages a collection of renderers and selects the appropriate one for each result type.

register(renderer, priority=0)[source]#

Parameters:

renderer (ResultRenderer) – ResultRenderer instance
priority (int) – Higher priority renderers are tried first

Return type:

register_for_type(result_type, renderer)[source]#

Parameters:

result_type (Type) – Python type to match
renderer (ResultRenderer) – ResultRenderer instance

Return type:

render(result, renderer_type=None)[source]#

Render a result using the appropriate renderer.

Parameters:

result (Any) – The function result to render
renderer_type (Optional[str]) – Optional specific renderer type to use

Return type:

Returns:

Rendered result dictionary

class uf.RjsfConfigBuilder[source]#

Bases: object

Builder for RJSF configurations with sensible defaults.

This class helps construct RJSF specifications by providing a fluent interface for configuring fields.

Example

>>> builder = RjsfConfigBuilder()
>>> builder.field('email', RjsfFieldConfig(format='email'))
>>> builder.field('message', RjsfFieldConfig(widget='textarea'))
>>> spec = builder.build(base_schema)

build(base_schema, base_ui_schema=None)[source]#

Build the final RJSF specification.

Parameters:

base_schema (dict) – Base JSON Schema to augment
base_ui_schema (Optional[dict]) – Optional base UI Schema to augment

Return type:

Returns:

Dictionary with ‘schema’ and ‘uiSchema’ keys

class_names(class_names)[source]#

Set CSS class names for the form.

Parameters:: class_names (str) – Space-separated CSS class names
Return type:: RjsfConfigBuilder
Returns:: Self for method chaining

field(param_name, config)[source]#

Configure a specific field.

Parameters:

param_name (str) – Name of the parameter/field
config (RjsfFieldConfig) – Configuration for the field

Return type:

RjsfConfigBuilder

Returns:

Self for method chaining

order(field_order)[source]#

Set the order of fields in the form.

Parameters:: field_order (list[str]) – List of field names in desired order
Return type:: RjsfConfigBuilder
Returns:: Self for method chaining

class uf.RjsfFieldConfig(widget=None, ui_options=<factory>, format=None, enum=None, description=None, placeholder=None, title=None, disabled=False, readonly=False, hidden=False, default=None)[source]#

Bases: object

Configuration for individual form fields.

This class allows fine-grained control over how form fields are rendered in the RJSF interface.

widget#: Widget type (e.g., ‘textarea’, ‘select’, ‘radio’, ‘date’)

ui_options#: Additional UI options for the widget

format#: JSON Schema format (e.g., ‘email’, ‘uri’, ‘date-time’)

enum#: List of allowed values (for dropdowns)

description#: Field description/help text

placeholder#: Placeholder text for the input

title#: Custom title for the field

disabled#: Whether the field is disabled

readonly#: Whether the field is read-only

hidden#: Whether to hide the field

default#: Default value for the field

Example

>>> email_config = RjsfFieldConfig(
...     widget='email',
...     format='email',
...     placeholder='user@example.com'
... )

to_json_schema_patch()[source]#

Convert to JSON Schema properties.

Return type:: dict
Returns:: Dictionary of JSON Schema properties to merge into schema

to_ui_schema_patch()[source]#

Convert to RJSF UI Schema properties.

Return type:: dict
Returns:: Dictionary of UI Schema properties for this field

class uf.SessionManager(secret_key, session_timeout=3600)[source]#

Bases: object

Manage user sessions.

Example

>>> sessions = SessionManager(secret_key='my-secret')
>>> session_id = sessions.create_session('admin')
>>> user = sessions.get_session(session_id)

cleanup_expired()[source]#

Remove expired sessions.

Return type:: int
Returns:: Number of sessions removed

create_session(username, data=None)[source]#

Create a new session.

Parameters:

username (str) – Username for session
data (Optional[dict]) – Optional session data

Return type:

str

Returns:

Session ID

delete_session(session_id)[source]#

Delete a session.

Parameters:: session_id (str) – Session ID
Return type:: bool
Returns:: True if deleted

get_session(session_id)[source]#

Get session data.

Parameters:: session_id (str) – Session ID
Return type:: Optional[dict]
Returns:: Session data or None if expired/invalid

class uf.Task(func, args=(), kwargs=None, task_id=None)[source]#

Bases: object

Represents a background task.

task_id#: Unique task identifier

func_name#: Name of the function

args#: Positional arguments

kwargs#: Keyword arguments

status#: Current task status

result#: Task result (if completed)

error#: Error message (if failed)

created_at#: When task was created

started_at#: When task started running

completed_at#: When task completed

progress#: Progress percentage (0-100)

execute()[source]#

Execute the task.

Return type:: Any
Returns:: Task result
Raises:: Exception – If task execution fails

to_dict()[source]#

Convert task to dictionary.

Return type:: dict
Returns:: Dictionary representation

class uf.TaskQueue(num_workers=1, max_queue_size=100)[source]#

Bases: object

FIFO queue for background tasks.

Example

>>> task_queue = TaskQueue(num_workers=2)
>>> task_queue.start()
>>> task_id = task_queue.submit(expensive_function, x=10, y=20)
>>> status = task_queue.get_status(task_id)
>>> result = task_queue.get_result(task_id)

cancel_task(task_id)[source]#

Cancel a task.

Parameters:: task_id (str) – Task ID
Return type:: bool
Returns:: True if cancelled

Note

Can only cancel pending tasks

get_result(task_id, wait=False, timeout=None)[source]#

Get task result.

Parameters:

task_id (str) – Task ID
wait (bool) – Whether to wait for completion
timeout (Optional[float]) – Optional timeout in seconds

Return type:

Returns:

Task result

Raises:

ValueError – If task not found
RuntimeError – If task failed
TimeoutError – If wait times out

get_status(task_id)[source]#

Get task status.

Parameters:: task_id (str) – Task ID
Return type:: Optional[TaskStatus]
Returns:: TaskStatus or None if not found

get_task(task_id)[source]#

Get task object.

Parameters:: task_id (str) – Task ID
Return type:: Optional[Task]
Returns:: Task object or None

queue_size()[source]#

Get current queue size.

Return type:: int
Returns:: Number of pending tasks

start()[source]#

Start worker threads.

Return type:: None

stats()[source]#

Get queue statistics.

Return type:: dict
Returns:: Dictionary with statistics

stop(wait=True)[source]#

Stop worker threads.

Parameters:: wait (bool) – Whether to wait for threads to finish
Return type:: None

submit(func, *args, task_id=None, **kwargs)[source]#

Submit a task for execution.

Parameters:

func (Callable) – Function to execute
*args – Positional arguments
task_id (Optional[str]) – Optional task ID
**kwargs – Keyword arguments

Return type:

str

Returns:

Task ID

Raises:

Full – If queue is full

class uf.TaskStatus(*values)[source]#

Bases: Enum

Status of a background task.

class uf.Theme(name, colors=<factory>, fonts=<factory>, css='')[source]#

Bases: object

UI theme configuration.

name#: Theme name

colors#: Color palette

fonts#: Font configuration

css#: Additional custom CSS

to_css()[source]#

Convert theme to CSS.

Return type:: str
Returns:: CSS string

class uf.ThemeConfig(default_theme='light', allow_toggle=True, available_themes=None, custom_theme=None)[source]#

Bases: object

Configuration for theme system.

Example

>>> config = ThemeConfig(
...     default_theme='dark',
...     allow_toggle=True,
...     available_themes=['light', 'dark', 'ocean']
... )

get_css()[source]#

Get CSS for all available themes.

Return type:: str
Returns:: Combined CSS string

get_js()[source]#

Get JavaScript for theme functionality.

Return type:: str
Returns:: JavaScript string

get_toggle_button_html()[source]#

Get HTML for theme toggle button.

Return type:: str
Returns:: HTML string or empty if toggle not allowed

class uf.UfApp(funcs, **mk_rjsf_app_kwargs)[source]#

Bases: object

Wrapper class for uf applications.

Provides a higher-level interface with additional conveniences beyond the raw qh app.

app#: The underlying qh/bottle/fastapi app

function_specs#: FunctionSpecStore for function metadata

funcs#: Dictionary mapping function names to callables

call(func_name, **kwargs)[source]#

Call a function directly by name.

Parameters:

func_name (str) – Name of the function to call
**kwargs – Arguments to pass to the function

Return type:

Returns:

Result of the function call

Raises:

KeyError – If function name not found

get_spec(func_name)[source]#

Get RJSF specification for a function.

Parameters:: func_name (str) – Name of the function
Return type:: dict
Returns:: Dictionary with schema and uiSchema
Raises:: KeyError – If function name not found

list_functions()[source]#

Get list of all function names.

Return type:: list[str]
Returns:: List of function name strings

run(host='localhost', port=8080, **kwargs)[source]#

Run the web application.

Parameters:

host (str) – Host to bind to
port (int) – Port to listen on
**kwargs – Additional arguments passed to app.run()

class uf.UfAppTester(app)[source]#

Bases: object

Context manager for testing uf apps.

Provides a testing context with common utilities and assertions.

Example

>>> with UfAppTester(app) as tester:
...     result = tester.submit_form('add', {'x': 10, 'y': 20})
...     tester.assert_success(result)
...     tester.assert_result_equals(result, 30)

assert_error_type(result, error_type)[source]#

Assert that error type matches.

Parameters:

result (dict) – Result dictionary
error_type (str) – Expected error type name

Raises:

AssertionError – If error types don’t match

assert_failure(result, message='')[source]#

Assert that a result indicates failure.

Parameters:

result (dict) – Result dictionary
message (str) – Optional custom error message

Raises:

AssertionError – If result is successful

assert_result_equals(result, expected)[source]#

Assert that result value equals expected.

Parameters:

result (dict) – Result dictionary
expected (Any) – Expected value

Raises:

AssertionError – If values don’t match

assert_success(result, message='')[source]#

Assert that a result indicates success.

Parameters:

result (dict) – Result dictionary
message (str) – Optional custom error message

Raises:

AssertionError – If result is not successful

get_history()[source]#

Get history of all results.

Return type:: list[dict]
Returns:: List of result dictionaries

submit_form(func_name, form_data)[source]#

Simulate form submission.

Parameters:

func_name (str) – Name of the function
form_data (dict) – Form data as dictionary

Return type:

Returns:

Result dictionary

class uf.UfTestClient(app)[source]#

Bases: object

Test client for uf applications.

Provides a convenient interface for testing uf apps without running a web server.

Example

>>> from uf import mk_rjsf_app
>>> app = mk_rjsf_app([my_func])
>>> client = UfTestClient(app)
>>> response = client.call_function('my_func', {'x': 10, 'y': 20})
>>> assert response['success']

call_function(func_name, params, *, expect_success=True)[source]#

Call a function with the given parameters.

Parameters:

func_name (str) – Name of the function to call
params (dict) – Dictionary of parameters
expect_success (bool) – Whether to expect success (raises on failure)

Return type:

Returns:

Result dictionary with ‘success’ and ‘result’ or ‘error’

Raises:

AssertionError – If expect_success=True and call fails

get_spec(func_name)[source]#

Get RJSF specification for a function.

Parameters:: func_name (str) – Name of the function
Return type:: dict
Returns:: Function specification dict
Raises:: KeyError – If function not found

list_functions()[source]#

Get list of available function names.

Return type:: list[str]
Returns:: List of function name strings

validate_params(func_name, params)[source]#

Validate parameters against function schema.

Parameters:

func_name (str) – Name of the function
params (dict) – Parameters to validate

Return type:

tuple[bool, Optional[str]]

Returns:

Tuple of (is_valid, error_message)

class uf.User(username, password_hash, roles=<factory>, permissions=<factory>, metadata=<factory>, created_at=<factory>, is_active=True)[source]#

Bases: object

User account information.

username#: Unique username

password_hash#: Hashed password

roles#: List of role names

permissions#: List of permission strings

metadata#: Additional user metadata

created_at#: When user was created

is_active#: Whether account is active

has_all_roles(roles)[source]#

Check if user has all of the given roles.

Return type:: bool

has_any_role(roles)[source]#

Check if user has any of the given roles.

Return type:: bool

has_permission(permission)[source]#

Check if user has a permission.

Return type:: bool

has_role(role)[source]#

Check if user has a role.

Return type:: bool

class uf.WebhookClient(url, headers=None, timeout=10.0, retry_count=3)[source]#

Bases: object

Client for sending webhooks.

Example

>>> client = WebhookClient('https://example.com/webhook')
>>> client.send(WebhookEvent('success', 'my_func', {}, result=42))

send(event, async_send=True)[source]#

Send a webhook event.

Parameters:

event (WebhookEvent) – WebhookEvent to send
async_send (bool) – Whether to send asynchronously

Return type:

Returns:

True if sent successfully (for sync sends)

class uf.WebhookEvent(event_type, func_name, params, result=None, error=None)[source]#

Bases: object

Represents a webhook event.

event_type#: Type of event (‘success’, ‘failure’, ‘start’)

func_name#: Name of the function

params#: Function parameters

result#: Function result (if success)

error#: Error message (if failure)

timestamp#: When event occurred

to_dict()[source]#

Convert to dictionary for JSON serialization.

Return type:: dict
Returns:: Dictionary representation

class uf.WebhookManager[source]#

Bases: object

Manage webhooks for multiple URLs and events.

Example

>>> manager = WebhookManager()
>>> manager.add_webhook('https://example.com/hook1', events=['success'])
>>> manager.add_webhook('https://example.com/hook2', events=['failure'])
>>> manager.trigger('success', 'my_func', {}, result=42)

add_webhook(url, events=None, headers=None, condition=None)[source]#

Add a webhook.

Parameters:

url (str) – Webhook URL
events (Optional[list[str]]) – List of event types to trigger on (None = all)
headers (Optional[dict]) – Optional HTTP headers
condition (Optional[Callable]) – Optional callable to filter events

Return type:

list_webhooks()[source]#

List all registered webhooks.

Return type:: list[dict]
Returns:: List of webhook configurations

remove_webhook(url)[source]#

Remove a webhook by URL.

Parameters:: url (str) – Webhook URL
Return type:: bool
Returns:: True if removed

trigger(event_type, func_name, params, result=None, error=None)[source]#

Trigger webhooks for an event.

Parameters:

event_type (str) – Event type
func_name (str) – Function name
params (dict) – Function parameters
result (Any) – Function result
error (Optional[str]) – Error message

Return type:

int

Returns:

Number of webhooks triggered

uf.add_openapi_routes(app, funcs, **spec_kwargs)[source]#

Add OpenAPI routes to an app.

Adds:

/openapi.json: OpenAPI specification
/docs: Swagger UI
/redoc: ReDoc UI

Parameters:

app (Any) – Web application
funcs (list[Callable]) – List of functions
**spec_kwargs – Arguments for generate_openapi_spec

Return type:

uf.apply_field_configs(schema, ui_schema, field_configs)[source]#

Apply field configurations to existing schemas.

Parameters:

schema (dict) – JSON Schema to modify
ui_schema (dict) – UI Schema to modify
field_configs (dict[str, RjsfFieldConfig]) – Mapping of field names to configurations

Return type:

tuple[dict, dict]

Returns:

Tuple of (modified_schema, modified_ui_schema)

Example

>>> configs = {
...     'email': get_field_config('email'),
...     'bio': get_field_config('multiline_text'),
... }
>>> schema, ui_schema = apply_field_configs(schema, ui_schema, configs)

uf.assert_field_required(spec, field_name)[source]#

Assert that a field is required.

Parameters:

spec (dict) – The specification
field_name (str) – Name of the field

Raises:

AssertionError – If field is not required

uf.assert_field_type(spec, field_name, expected_type)[source]#

Assert that a field has the expected type.

Parameters:

spec (dict) – The specification
field_name (str) – Name of the field
expected_type (str) – Expected JSON Schema type

Raises:

AssertionError – If type doesn’t match

uf.assert_has_field(spec, field_name)[source]#

Assert that a spec has a specific field.

Parameters:

spec (dict) – The specification
field_name (str) – Name of the field to check

Raises:

AssertionError – If field not found

uf.assert_valid_rjsf_spec(spec)[source]#

Assert that a specification is valid RJSF format.

Parameters:: spec (dict) – The specification to validate
Raises:: AssertionError – If spec is invalid

uf.async_to_sync(async_func)[source]#

Convert an async function to a synchronous wrapper.

Parameters:: async_func (Callable) – Async function to wrap
Return type:: Callable
Returns:: Synchronous wrapper function

Example

>>> async def fetch_data(url: str):
...     # async operations
...     return data
>>> sync_fetch = async_to_sync(fetch_data)
>>> result = sync_fetch('https://example.com')

uf.auto_group_by_module(funcs)[source]#

Automatically group functions by their module.

Parameters:: funcs (Iterable[Callable]) – Functions to organize
Return type:: FunctionOrganizer
Returns:: FunctionOrganizer with module-based groups (a functools.partial goes with the module of the function it wraps)

Example

>>> from myapp import user_ops, report_ops
>>> funcs = [user_ops.create, user_ops.delete, report_ops.generate]
>>> organizer = auto_group_by_module(funcs)

uf.auto_group_by_prefix(funcs, separator='_')[source]#

Automatically group functions by name prefix.

Groups functions based on the part of their name before the separator. For example, with separator=”_”:

user_create, user_delete → “user” group
report_generate, report_export → “report” group

Parameters:

funcs (Iterable[Callable]) – Functions to organize
separator (str) – Separator character (default: “_”)

Return type:

FunctionOrganizer

Returns:

FunctionOrganizer with auto-generated groups

Example

>>> funcs = [user_create, user_delete, report_generate, admin_reset]
>>> organizer = auto_group_by_prefix(funcs)
>>> # Creates groups: 'user', 'report', 'admin'

uf.auto_group_by_tag(funcs, tag_attr='__uf_group__')[source]#

Automatically group functions by a tag attribute.

Functions can be tagged with a group name using an attribute.

Parameters:

funcs (Iterable[Callable]) – Functions to organize
tag_attr (str) – Name of the attribute to use for grouping

Return type:

FunctionOrganizer

Returns:

FunctionOrganizer with tag-based groups

Example

>>> def create_user(name: str):
...     pass
>>> create_user.__uf_group__ = 'Admin'
>>>
>>> organizer = auto_group_by_tag([create_user, other_func])

uf.background(queue_name='default', task_queue=None)[source]#

Decorator to run function in background.

Parameters:

queue_name (str) – Name of the queue to use
task_queue (Optional[TaskQueue]) – Optional TaskQueue instance

Returns:

Decorator function

Example

>>> @background()
... def send_email(to: str, subject: str):
...     # Long-running email sending
...     pass
>>>
>>> task_id = send_email('user@example.com', 'Hello')
>>> # Returns immediately with task_id

uf.cached(ttl=3600, backend=None, key_func=None)[source]#

Decorator to cache function results.

Parameters:

ttl (int) – Time to live in seconds
backend (Optional[CacheBackend]) – Cache backend (uses global MemoryCache if None)
key_func (Optional[Callable]) – Optional function to generate cache key

Returns:

Decorator function

Example

>>> @cached(ttl=3600)
... def expensive_calculation(x: int, y: int) -> int:
...     # Only runs once per unique (x, y) combination
...     return heavy_computation(x, y)
>>>
>>> result = expensive_calculation(10, 20)  # Computes
>>> result2 = expensive_calculation(10, 20)  # From cache

uf.deprecated(message=None)[source]#

Decorator to mark a function as deprecated.

Parameters:: message (Optional[str]) – Optional deprecation message
Returns:: Decorator function

Example

>>> @deprecated("Use new_function instead")
... def old_function():
...     pass

uf.dict_to_pydantic(data, model_class)[source]#

Convert dictionary to Pydantic model instance.

Parameters:

data (dict) – Dictionary of data
model_class – Pydantic model class

Return type:

Returns:

Pydantic model instance

Raises:

ValidationError – If data doesn’t match model schema

uf.enable_history(func, max_size=100)[source]#

Decorator to enable call history for a function.

Parameters:: max_size (int) – Maximum history size
Return type:: Callable
Returns:: Decorator function

Example

>>> @enable_history
... def my_function(x: int):
...     return x * 2

uf.field_config(**field_configs)[source]#

Decorator to configure specific fields of a function.

Parameters:: **field_configs (RjsfFieldConfig) – Keyword arguments mapping parameter names to configs
Returns:: Decorator function

Example

>>> from uf.rjsf_config import get_field_config
>>>
>>> @field_config(
...     email=get_field_config('email'),
...     bio=get_field_config('multiline_text')
... )
... def create_profile(email: str, bio: str):
...     pass

uf.function_uses_pydantic(func)[source]#

Check if a function uses Pydantic models in its signature.

Parameters:: func (Callable) – Function to check
Return type:: bool
Returns:: True if any parameter is a Pydantic model

uf.generate_openapi_spec(funcs, title='API', version='1.0.0', description='', servers=None)[source]#

Generate OpenAPI 3.0 specification.

Parameters:

funcs (list[Callable]) – List of functions
title (str) – API title
version (str) – API version
description (str) – API description
servers (Optional[list[dict]]) – Optional list of server configs

Return type:

Returns:

OpenAPI specification dictionary

uf.get_example(func)[source]#

Get example arguments from a function.

Parameters:: func (Callable) – Function to get example from
Return type:: Optional[tuple[tuple, dict]]
Returns:: Tuple of (args, kwargs) or None if no example

Example

>>> example = get_example(my_function)
>>> if example:
...     args, kwargs = example
...     result = my_function(*args, **kwargs)

uf.get_field_config(config_name)[source]#

Get a predefined field configuration by name.

Parameters:: config_name (str) – Name of the configuration (e.g., ‘email’, ‘multiline_text’)
Return type:: RjsfFieldConfig
Returns:: RjsfFieldConfig instance
Raises:: KeyError – If config_name not found

Example

>>> email_config = get_field_config('email')
>>> email_config.format
'email'

uf.get_field_configs(func)[source]#

Get field configurations from a function.

Parameters:: func (Callable) – Function to get field configs from
Return type:: dict[str, RjsfFieldConfig]
Returns:: Dictionary mapping parameter names to RjsfFieldConfig

Example

>>> field_configs = get_field_configs(my_function)
>>> if 'email' in field_configs:
...     print(field_configs['email'].format)

uf.get_field_dependencies(func)[source]#

Get field dependencies from a function.

Parameters:: func (Callable) – Function to get dependencies from
Return type:: list[FieldDependency]
Returns:: List of FieldDependency objects

uf.get_global_auth_backend()[source]#

Get the global authentication backend.

Return type:: Optional[AuthBackend]

uf.get_global_cache_backend()[source]#

Get the global cache backend.

Return type:: Optional[CacheBackend]

uf.get_global_history_manager()[source]#

Get the global history manager.

Return type:: HistoryManager
Returns:: Global HistoryManager instance

uf.get_global_registry()[source]#

Get the global transformation registry.

Return type:: InputTransformRegistry
Returns:: The global InputTransformRegistry instance

uf.get_global_renderer_registry()[source]#

Get the global renderer registry.

Return type:: ResultRendererRegistry
Returns:: The global ResultRendererRegistry instance

uf.get_global_task_queue(name='default')[source]#

Get a global task queue by name.

Parameters:: name (str) – Queue name
Return type:: Optional[TaskQueue]
Returns:: TaskQueue or None

uf.get_global_webhook_manager()[source]#

Get the global webhook manager.

Return type:: Optional[WebhookManager]

uf.get_group(func)[source]#

Get group name from a function.

Parameters:: func (Callable) – Function to get group from
Return type:: Optional[str]
Returns:: Group name or None

Example

>>> group = get_group(my_function)
>>> if group:
...     print(f"Function is in group: {group}")

uf.get_result_renderer(func)[source]#

Get the result renderer type for a function.

Parameters:: func (Callable) – Function to check
Return type:: Optional[str]
Returns:: Renderer type string or None

uf.get_theme(name)[source]#

Get a built-in theme by name.

Parameters:: name (str) – Theme name
Return type:: Optional[Theme]
Returns:: Theme object or None

Example

>>> theme = get_theme('dark')
>>> css = theme.to_css()

uf.get_ui_config(func)[source]#

Get UI configuration from a function.

Parameters:: func (Callable) – Function to get config from
Return type:: Optional[dict]
Returns:: Configuration dictionary or None if not configured

Example

>>> config = get_ui_config(my_function)
>>> if config:
...     print(config['title'])

uf.group(group_name)[source]#

Decorator to assign a function to a group.

Simpler alternative to ui_config when you only need to set the group.

Parameters:: group_name (str) – Name of the group
Returns:: Decorator function

Example

>>> @group("Admin")
... def delete_user(user_id: int):
...     pass

uf.hidden(func)[source]#

Decorator to hide a function from the UI.

The function will still be callable via the API but won’t appear in the UI navigation.

Parameters:: func (Callable) – Function to hide
Return type:: Callable
Returns:: Decorated function

Example

>>> @hidden
... def internal_function():
...     pass

uf.is_async_function(func)[source]#

Check if a function is async.

Parameters:: func (Callable) – Function to check
Return type:: bool
Returns:: True if function is async

Example

>>> async def my_async_func():
...     pass
>>> is_async_function(my_async_func)
True

uf.is_hidden(func)[source]#

Check if a function is hidden from UI.

Parameters:: func (Callable) – Function to check
Return type:: bool
Returns:: True if function is hidden, False otherwise

Example

>>> if not is_hidden(my_function):
...     # Show in UI
...     pass

uf.is_pydantic_model(obj)[source]#

Check if an object is a Pydantic model.

Parameters:: obj (Any) – Object to check
Return type:: bool
Returns:: True if object is a Pydantic BaseModel

uf.make_sync_compatible(funcs)[source]#

Convert any async functions in the list to sync wrappers.

Parameters:: funcs (list[Callable]) – List of functions (may include async)
Return type:: list[Callable]
Returns:: List of functions (all synchronous)

Example

>>> funcs = [sync_func, async_func, another_sync]
>>> compatible = make_sync_compatible(funcs)
>>> # All functions in compatible are now callable synchronously

uf.mk_grouped_app(groups, **mk_rjsf_app_kwargs)[source]#

Create a uf app with grouped function navigation.

Parameters:

groups (Iterable[FunctionGroup]) – Iterable of FunctionGroup objects
**mk_rjsf_app_kwargs – Arguments passed to mk_rjsf_app

Returns:

Configured web application with grouped navigation

Example

>>> admin_group = FunctionGroup('Admin', [user_create, user_delete])
>>> reports_group = FunctionGroup('Reports', [generate_report])
>>> app = mk_grouped_app([admin_group, reports_group])

uf.mk_rjsf_app(funcs, *, config=None, input_trans=None, output_trans=None, rjsf_config=None, ui_schema_factory=None, param_to_prop_type=None, page_title='Function Interface', function_display_names=None, custom_css=None, rjsf_theme='default', add_ui=True, **qh_kwargs)[source]#

Create an RJSF-backed web app from a list of functions.

This is the main entry point for uf. It combines qh’s HTTP service generation with ju’s RJSF form generation to create a complete web interface for your functions.

Parameters:

funcs (Union[Iterable[Callable], Mapping[str, Callable]]) – Iterable of callables to expose via web UI – functions, partials, callable instances (named by uf.specs.name_of_func) – or a mapping {name: callable} to name them yourself. The name is the route and the key of the form spec
config (Optional[Any]) – Optional qh.AppConfig for HTTP service configuration
input_trans (Optional[Callable]) – Optional input transformation function for qh
output_trans (Optional[Callable]) – Optional output transformation function for qh
rjsf_config (Optional[dict]) – Optional configuration dict for RJSF generation
ui_schema_factory (Optional[Callable]) – Optional callable to customize UI schema
param_to_prop_type (Optional[Callable]) – Optional callable to map parameters to types
page_title (str) – Title for the web interface
function_display_names (Optional[Mapping]) – Optional mapping to override function names
custom_css (Optional[str]) – Optional custom CSS for the web interface
rjsf_theme (str) – RJSF theme to use (‘default’, ‘material-ui’, etc.)
add_ui (bool) – Whether to add UI routes (set False for API-only)
**qh_kwargs – Additional keyword arguments passed to qh.mk_app

Returns:

HTTP endpoints for each function
RJSF-based web interface (if add_ui=True)
API routes for function specs

Return type:

A configured web application (bottle or FastAPI app) with

Example

>>> def add(x: int, y: int) -> int:
...     '''Add two numbers'''
...     return x + y
...
>>> def greet(name: str) -> str:
...     '''Greet a person'''
...     return f"Hello, {name}!"
...
>>> app = mk_rjsf_app([add, greet])
>>> # app.run()  # Start the web server

Example with customization:

>>> from uf import mk_rjsf_app, RjsfFieldConfig
>>>
>>> def send_email(to: str, subject: str, body: str):
...     '''Send an email'''
...     pass
...
>>> app = mk_rjsf_app(
...     [send_email],
...     page_title="Email Sender",
...     custom_css="body { background: #f0f0f0; }",
... )

uf.pydantic_model_to_json_schema(model_class)[source]#

Convert a Pydantic model to JSON Schema.

Parameters:: model_class – Pydantic model class
Return type:: dict
Returns:: JSON Schema dictionary

Example

>>> from pydantic import BaseModel
>>> class User(BaseModel):
...     name: str
...     age: int
>>> schema = pydantic_model_to_json_schema(User)

uf.pydantic_to_dict(obj)[source]#

Convert Pydantic model instance to dictionary.

Parameters:: obj (Any) – Pydantic model instance
Return type:: dict
Returns:: Dictionary representation

uf.rate_limit(calls, period)[source]#

Decorator to mark a function with rate limiting metadata.

This is metadata-only; actual rate limiting must be implemented separately.

Parameters:

calls (int) – Number of calls allowed
period (int) – Time period in seconds

Returns:

Decorator function

Example

>>> @rate_limit(calls=10, period=60)  # 10 calls per minute
... def send_email(to: str, subject: str):
...     pass

uf.register_renderer(renderer, priority=0)[source]#

Parameters:

renderer (ResultRenderer) – ResultRenderer instance
priority (int) – Higher priority renderers are tried first

Return type:

uf.register_type(*args, **kwargs)[source]#

This is a convenience function that uses the global registry. See InputTransformRegistry.register_type for full documentation.

uf.render_result(result, renderer_type=None)[source]#

Render a result using the global registry.

Parameters:

result (Any) – The function result to render
renderer_type (Optional[str]) – Optional specific renderer type to use

Return type:

Returns:

Rendered result dictionary

uf.require_auth(backend, roles=None, permissions=None)[source]#

Decorator to require authentication for a function.

Parameters:

backend (AuthBackend) – Authentication backend
roles (Optional[list[str]]) – Required roles (any)
permissions (Optional[list[str]]) – Required permissions (all)

Returns:

Decorator function

Example

>>> backend = DictAuthBackend.from_dict({
...     'admin': {'password': 'secret', 'roles': ['admin']}
... })
>>> @require_auth(backend, roles=['admin'])
... def delete_all():
...     pass

uf.requires_auth(*, roles=None, permissions=None)[source]#

Decorator to mark a function as requiring authentication.

This is metadata-only; actual authentication must be implemented separately in the application layer.

Parameters:

roles (Optional[list[str]]) – List of required roles
permissions (Optional[list[str]]) – List of required permissions

Returns:

Decorator function

Example

>>> @requires_auth(roles=['admin'], permissions=['user:delete'])
... def delete_user(user_id: int):
...     pass

uf.result_renderer(renderer_type)[source]#

Decorator to specify result renderer for a function.

Parameters:: renderer_type (str) – Type of renderer to use
Returns:: Decorator function

Example

>>> @result_renderer('table')
... def get_users() -> list[dict]:
...     return [{'name': 'Alice', 'age': 30}]

uf.retry_async(max_attempts=3, delay=1.0)[source]#

Decorator to add retry logic to async functions.

Parameters:

max_attempts (int) – Maximum number of attempts
delay (float) – Delay between attempts in seconds

Returns:

Decorator function

Example

>>> @retry_async(max_attempts=3, delay=2.0)
... async def unreliable_api_call():
...     # May fail, will retry
...     return await external_api.fetch()

uf.test_ui_function(func, test_inputs, *, expected_output=None, expected_exception=None)[source]#

Test a function with form-like input.

Parameters:

func (Callable) – Function to test
test_inputs (dict) – Dictionary of test parameters
expected_output (Any) – Expected return value (if any)
expected_exception (Optional[type]) – Expected exception type (if any)

Return type: