uf
uf - UI Fast: Minimal-boilerplate web UIs for Python functions.
uf bridges functions to HTTP services (via qh) to Web UI forms (via ju.rjsf), following the “convention over configuration” philosophy.
- Basic usage:
>>> from uf import mk_rjsf_app >>> >>> def add(x: int, y: int) -> int: ... '''Add two numbers''' ... return x + y >>> >>> app = mk_rjsf_app([add]) >>> # app.run() # Start the web server
The main entry points are: - mk_rjsf_app: Create a web app from functions (functional interface) - UfApp: Object-oriented wrapper with additional conveniences - FunctionSpecStore: Manage function specifications (advanced usage)
Advanced features: - ui_config: Decorator for UI metadata - RjsfFieldConfig: Field configuration class - FunctionGroup: Group functions for organization - InputTransformRegistry: Custom type transformations - Field dependencies and interactions - Testing utilities
- class uf.ApiKey(key: str, name: str, permissions: list[str] | None = None, expires_at: datetime | None = None)[source]
API key for programmatic access.
- key
The API key string
- name
Descriptive name
- permissions
List of allowed permissions
- created_at
When key was created
- expires_at
Optional expiration
- is_active
Whether key is active
- class uf.ApiKeyManager(key_prefix: str = 'sk_')[source]
Manage API keys for programmatic access.
Example
>>> api_keys = ApiKeyManager() >>> key = api_keys.create_key('mobile_app', permissions=['read']) >>> print(f"Your API key: {key.key}") >>> # Later, validate >>> if api_keys.validate_key(key.key, 'read'): ... # Allow access
- create_key(name: str, permissions: list[str] | None = None, expires_in_days: int | None = None) ApiKey[source]
Create a new API key.
- Parameters:
name – Descriptive name for the key
permissions – List of allowed permissions
expires_in_days – Optional expiration in days
- Returns:
Created ApiKey object
- class uf.AsyncFunctionWrapper(func: Callable)[source]
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)
- class uf.AuthBackend[source]
Base authentication backend.
Subclass this to create custom authentication backends.
- authenticate(username: str, password: str) User | None[source]
Authenticate a user.
- Parameters:
username – Username
password – Password
- Returns:
User object if authenticated, None otherwise
- create_user(username: str, password: str, roles: list[str] | None = None, permissions: list[str] | None = None, **metadata) User[source]
Create a new user.
- Parameters:
username – Username
password – Plain text password
roles – List of roles
permissions – List of permissions
**metadata – Additional user metadata
- Returns:
Created User object
- delete_user(username: str) bool[source]
Delete a user.
- Parameters:
username – Username
- Returns:
True if deleted successfully
- class uf.CacheBackend[source]
Base class for cache backends.
- delete(key: str) bool[source]
Delete a key from cache.
- Parameters:
key – Cache key
- Returns:
True if deleted
- exists(key: str) bool[source]
Check if key exists in cache.
- Parameters:
key – Cache key
- Returns:
True if exists and not expired
- class uf.CallHistory(max_size: int = 100)[source]
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: str | None = None) None[source]
Clear history.
- Parameters:
func_name – Function to clear, or None for all
- classmethod from_dict(data: dict, max_size: int = 100) CallHistory[source]
Create from dictionary.
- Parameters:
data – Dictionary with history data
max_size – Maximum size per function
- Returns:
CallHistory instance
- get_recent(func_name: str, limit: int = 10) list[FunctionCall][source]
Get recent calls for a function.
- Parameters:
func_name – Function name
limit – Maximum number to return
- Returns:
List of recent FunctionCall objects (newest first)
- get_successful_calls(func_name: str, limit: int = 10) list[FunctionCall][source]
Get recent successful calls.
- Parameters:
func_name – Function name
limit – Maximum number to return
- Returns:
List of successful FunctionCall objects
- record(func_name: str, params: dict, result: Any = None, success: bool = True, error: str | None = None) None[source]
Record a function call.
- Parameters:
func_name – Name of the function
params – Parameters used
result – Result returned
success – Whether call succeeded
error – Error message if failed
- class uf.ConditionalFieldConfig(field_name: str, *, condition: dict, then_schema: dict | None = None, else_schema: dict | None = None)[source]
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'}} ... )
- class uf.DependencyBuilder[source]
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() list[FieldDependency][source]
Build and return the list of dependencies.
- Returns:
List of FieldDependency objects
- custom(condition: Callable[[Any], bool]) DependencyBuilder[source]
Condition: custom callable.
- Parameters:
condition – Function that takes value and returns bool
- Returns:
Self for method chaining
- disable(target_field: str) DependencyBuilder[source]
Action: disable target field.
- Parameters:
target_field – Name of field to disable
- Returns:
Self for method chaining
- enable(target_field: str) DependencyBuilder[source]
Action: enable target field.
- Parameters:
target_field – Name of field to enable
- Returns:
Self for method chaining
- equals(value: Any) DependencyBuilder[source]
Condition: field equals value.
- Parameters:
value – Value to compare against
- Returns:
Self for method chaining
- greater_than(value: Any) DependencyBuilder[source]
Condition: field is greater than value.
- Parameters:
value – Value to compare against
- Returns:
Self for method chaining
- hide(target_field: str) DependencyBuilder[source]
Action: hide target field.
- Parameters:
target_field – Name of field to hide
- Returns:
Self for method chaining
- in_list(values: list) DependencyBuilder[source]
Condition: field value is in list.
- Parameters:
values – List of acceptable values
- Returns:
Self for method chaining
- is_falsy() DependencyBuilder[source]
Condition: field is falsy.
- Returns:
Self for method chaining
- is_truthy() DependencyBuilder[source]
Condition: field is truthy.
- Returns:
Self for method chaining
- less_than(value: Any) DependencyBuilder[source]
Condition: field is less than value.
- Parameters:
value – Value to compare against
- Returns:
Self for method chaining
- make_optional(target_field: str) DependencyBuilder[source]
Action: make target field optional.
- Parameters:
target_field – Name of field to make optional
- Returns:
Self for method chaining
- not_equals(value: Any) DependencyBuilder[source]
Condition: field does not equal value.
- Parameters:
value – Value to compare against
- Returns:
Self for method chaining
- require(target_field: str) DependencyBuilder[source]
Action: make target field required.
- Parameters:
target_field – Name of field to require
- Returns:
Self for method chaining
- show(target_field: str) DependencyBuilder[source]
Action: show target field.
- Parameters:
target_field – Name of field to show
- Returns:
Self for method chaining
- when(field_name: str) DependencyBuilder[source]
Start a dependency condition on a field.
- Parameters:
field_name – Name of the source field
- Returns:
Self for method chaining
- class uf.DictAuthBackend[source]
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')
- create_user(username: str, password: str, roles: list[str] | None = None, permissions: list[str] | None = None, **metadata) User[source]
Create a new user.
- classmethod from_dict(users_data: dict) DictAuthBackend[source]
Create backend from dictionary.
- Parameters:
users_data – Dictionary mapping usernames to user info
Example
>>> backend = DictAuthBackend.from_dict({ ... 'admin': {'password': 'secret', 'roles': ['admin']}, ... 'user': {'password': 'pass', 'roles': ['user']}, ... })
- class uf.DiskCache(cache_dir: str = '.uf_cache', default_ttl: int = 3600)[source]
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)
- class uf.FieldDependency(source_field: str, target_field: str, condition: Callable[[Any], bool], action: DependencyAction, else_action: DependencyAction | None = None)[source]
Define a dependency between form fields.
- source_field
Name of the field that triggers the dependency
- Type:
str
- target_field
Name of the field affected by the dependency
- Type:
str
- condition
Function that takes source value and returns bool
- Type:
Callable[[Any], bool]
- action
Action to perform when condition is true
- else_action
Optional action when condition is false
- Type:
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: Any) DependencyAction[source]
Check the condition and return the appropriate action.
- Parameters:
value – Value of the source field
- Returns:
The action to perform
- class uf.FormDataBuilder[source]
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() ... )
- field(name: str, value: Any) FormDataBuilder[source]
Add a field to the form data.
- Parameters:
name – Field name
value – Field value
- Returns:
Self for method chaining
- fields(**kwargs) FormDataBuilder[source]
Add multiple fields at once.
- Parameters:
**kwargs – Field name-value pairs
- Returns:
Self for method chaining
- class uf.FunctionCall(func_name: str, params: dict, result: Any = None, success: bool = True, error: str | None = None)[source]
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: dict) FunctionCall[source]
Create from dictionary.
- Parameters:
data – Dictionary with call data
- Returns:
FunctionCall instance
- class uf.FunctionGroup(name: str, funcs: list[Callable] = <factory>, description: str = '', icon: str | None = None, order: int = 0, collapsed: bool = False)[source]
Group of functions with metadata.
- name
Name of the group
- Type:
str
- funcs
Functions in this group
- Type:
list[Callable]
- description
Description of the group
- Type:
str
- icon
Optional icon identifier for the group
- Type:
str | None
- order
Display order (lower numbers first)
- Type:
int
- collapsed
Whether the group starts collapsed in UI
- Type:
bool
Example
>>> math_funcs = FunctionGroup( ... 'Math', ... [add, subtract, multiply, divide], ... description='Mathematical operations', ... icon='calculator' ... )
- add_function(func: Callable) FunctionGroup[source]
Add a function to this group.
- Parameters:
func – Function to add
- Returns:
Self for method chaining
- class uf.FunctionOrganizer[source]
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: str, func: Callable) FunctionOrganizer[source]
Add a function to an existing group.
- Parameters:
group_name – Name of the group
func – Function to add
- Returns:
Self for method chaining
- Raises:
ValueError – If group doesn’t exist
- add_ungrouped(func: Callable) FunctionOrganizer[source]
Add a function without a group.
- Parameters:
func – Function to add
- Returns:
Self for method chaining
- get_all_functions() list[Callable][source]
Get all functions across all groups.
- Returns:
List of all functions
- get_groups() list[FunctionGroup][source]
Get all groups, sorted by order.
- Returns:
List of FunctionGroup objects sorted by order
- group(name: str, funcs: Iterable[Callable] | None = None, *, description: str = '', icon: str | None = None, order: int = 0, collapsed: bool = False) FunctionGroup[source]
Create and add a function group.
- Parameters:
name – Name of the group
funcs – Optional functions to add to group
description – Description of the group
icon – Optional icon identifier
order – Display order
collapsed – Whether to start collapsed
- Returns:
The created FunctionGroup
Example
>>> organizer.group( ... 'Database', ... [save_record, load_record], ... description='Database operations', ... icon='database' ... )
- class uf.FunctionSpecStore(funcs: Iterable[Callable], *, rjsf_config: dict | None = None, ui_schema_factory: Callable | None = None, param_to_prop_type: Callable | None = None)[source]
A mapping from function names to their RJSF specifications.
Lazily generates and caches form specs for each function.
- Parameters:
funcs – Iterable of callable functions to generate specs for
rjsf_config – Optional configuration dict for RJSF generation
ui_schema_factory – Optional callable to customize UI schema generation
param_to_prop_type – Optional callable to map parameters to property types
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
- class uf.HistoryManager(max_history: int = 100)[source]
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: str, func_name: str) Preset | None[source]
Get a preset.
- Parameters:
preset_name – Name of preset
func_name – Function name
- Returns:
Preset object or None
- get_presets(func_name: str) list[Preset][source]
Get all presets for a function.
- Parameters:
func_name – Function name
- Returns:
List of Preset objects
- get_recent_calls(func_name: str, limit: int = 10) list[FunctionCall][source]
Get recent calls.
- Parameters:
func_name – Function name
limit – Maximum number to return
- Returns:
List of recent FunctionCall objects
- classmethod load_from_file(filepath: str, max_history: int = 100) HistoryManager[source]
Load from file.
- Parameters:
filepath – Path to load from
max_history – Maximum history size
- Returns:
HistoryManager instance
- record_call(func_name: str, params: dict, result: Any = None, success: bool = True, error: str | None = None) None[source]
Record a function call.
- Parameters:
func_name – Function name
params – Parameters used
result – Result returned
success – Whether call succeeded
error – Error message if failed
- class uf.InputTransformRegistry[source]
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: Any, py_type: Type) Any[source]
Transform a JSON value to Python type.
- Parameters:
value – JSON value to transform
py_type – Target Python type
- Returns:
Transformed value
- get_all_registered_types() list[Type][source]
Get list of all registered types.
- Returns:
List of registered Python types
- get_handler(py_type: Type) dict | None[source]
Get handler for a type.
- Parameters:
py_type – Python type to look up
- Returns:
Handler dict or None if not registered
- get_ui_config(py_type: Type) RjsfFieldConfig | None[source]
Get UI configuration for a type.
- Parameters:
py_type – Python type to look up
- Returns:
RjsfFieldConfig or None if not registered
- mk_input_trans_for_funcs(funcs: list[Callable]) Callable[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 of functions to create transformation for
- 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() Callable[source]
Create output transformation for qh.
- 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: Type, *, to_json: Callable[[Any], Any] | None = None, from_json: Callable[[Any], Any] | None = None, ui_widget: str | None = None, ui_config: RjsfFieldConfig | None = None, json_schema_type: str | None = None, json_schema_format: str | None = None) None[source]
Register a type with both qh and UI configuration.
- Parameters:
py_type – Python type to register
to_json – Function to convert Python type to JSON-serializable
from_json – Function to convert JSON to Python type
ui_widget – RJSF widget to use for this type
ui_config – Full RjsfFieldConfig for this type
json_schema_type – JSON Schema type (e.g., ‘string’, ‘number’)
json_schema_format – JSON Schema format (e.g., ‘date-time’, ‘email’)
Example
>>> from pathlib import Path >>> registry.register_type( ... Path, ... to_json=str, ... from_json=Path, ... ui_widget='text', ... json_schema_type='string' ... )
- class uf.MemoryCache(default_ttl: int = 3600, max_size: int = 1000)[source]
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')
- class uf.OpenAPIConfig(title: str = 'API', version: str = '1.0.0', description: str = '', servers: list[dict] | None = None, enable_swagger: bool = True, enable_redoc: bool = True)[source]
Configuration for OpenAPI generation.
Example
>>> config = OpenAPIConfig( ... title="My API", ... version="2.0.0", ... description="API for my application" ... )
- class uf.Preset(name: str, func_name: str, params: dict, description: str = '')[source]
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
- class uf.PresetManager[source]
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: str, func_name: str) bool[source]
Delete a preset.
- Parameters:
preset_name – Name of preset
func_name – Function name
- Returns:
True if deleted, False if not found
- classmethod from_dict(data: dict) PresetManager[source]
Create from dictionary.
- Parameters:
data – Dictionary with preset data
- Returns:
PresetManager instance
- get(preset_name: str, func_name: str) Preset | None[source]
Get a preset.
- Parameters:
preset_name – Name of preset
func_name – Function name
- Returns:
Preset object or None
- list_presets(func_name: str) list[Preset][source]
List all presets for a function.
- Parameters:
func_name – Function name
- Returns:
List of Preset objects
- class uf.ResultRenderer[source]
Base class for result renderers.
Subclass this to create custom renderers for specific result types.
- can_render(result: Any) bool[source]
Check if this renderer can handle the given result.
- Parameters:
result – The function result to render
- Returns:
True if this renderer can handle the result
- render(result: Any) dict[source]
Render the result to a displayable format.
- Parameters:
result – The function result to render
- Returns:
‘type’: Renderer type (e.g., ‘table’, ‘chart’, ‘json’)
’data’: Rendered data
’options’: Optional rendering options
- Return type:
Dictionary with rendering information
- class uf.ResultRendererRegistry[source]
Registry for result renderers.
Manages a collection of renderers and selects the appropriate one for each result type.
- register(renderer: ResultRenderer, priority: int = 0) None[source]
Register a renderer.
- Parameters:
renderer – ResultRenderer instance
priority – Higher priority renderers are tried first
- register_for_type(result_type: Type, renderer: ResultRenderer) None[source]
Register a renderer for a specific type.
- Parameters:
result_type – Python type to match
renderer – ResultRenderer instance
- class uf.RjsfConfigBuilder[source]
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: dict, base_ui_schema: dict | None = None) dict[source]
Build the final RJSF specification.
- Parameters:
base_schema – Base JSON Schema to augment
base_ui_schema – Optional base UI Schema to augment
- Returns:
Dictionary with ‘schema’ and ‘uiSchema’ keys
- class_names(class_names: str) RjsfConfigBuilder[source]
Set CSS class names for the form.
- Parameters:
class_names – Space-separated CSS class names
- Returns:
Self for method chaining
- field(param_name: str, config: RjsfFieldConfig) RjsfConfigBuilder[source]
Configure a specific field.
- Parameters:
param_name – Name of the parameter/field
config – Configuration for the field
- Returns:
Self for method chaining
- order(field_order: list[str]) RjsfConfigBuilder[source]
Set the order of fields in the form.
- Parameters:
field_order – List of field names in desired order
- Returns:
Self for method chaining
- class uf.RjsfFieldConfig(widget: str | None = None, ui_options: dict = <factory>, format: str | None = None, enum: list | None = None, description: str | None = None, placeholder: str | None = None, title: str | None = None, disabled: bool = False, readonly: bool = False, hidden: bool = False, default: Any | None = None)[source]
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’)
- Type:
str | None
- ui_options
Additional UI options for the widget
- Type:
dict
- format
JSON Schema format (e.g., ‘email’, ‘uri’, ‘date-time’)
- Type:
str | None
- enum
List of allowed values (for dropdowns)
- Type:
list | None
- description
Field description/help text
- Type:
str | None
- placeholder
Placeholder text for the input
- Type:
str | None
- title
Custom title for the field
- Type:
str | None
- disabled
Whether the field is disabled
- Type:
bool
- readonly
Whether the field is read-only
- Type:
bool
Whether to hide the field
- Type:
bool
- default
Default value for the field
- Type:
Any | None
Example
>>> email_config = RjsfFieldConfig( ... widget='email', ... format='email', ... placeholder='user@example.com' ... )
- class uf.SessionManager(secret_key: str, session_timeout: int = 3600)[source]
Manage user sessions.
Example
>>> sessions = SessionManager(secret_key='my-secret') >>> session_id = sessions.create_session('admin') >>> user = sessions.get_session(session_id)
- create_session(username: str, data: dict | None = None) str[source]
Create a new session.
- Parameters:
username – Username for session
data – Optional session data
- Returns:
Session ID
- class uf.Task(func: Callable, args: tuple = (), kwargs: dict | None = None, task_id: str | None = None)[source]
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)
- class uf.TaskQueue(num_workers: int = 1, max_queue_size: int = 100)[source]
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: str) bool[source]
Cancel a task.
- Parameters:
task_id – Task ID
- Returns:
True if cancelled
Note
Can only cancel pending tasks
- get_result(task_id: str, wait: bool = False, timeout: float | None = None) Any[source]
Get task result.
- Parameters:
task_id – Task ID
wait – Whether to wait for completion
timeout – Optional timeout in seconds
- Returns:
Task result
- Raises:
ValueError – If task not found
RuntimeError – If task failed
TimeoutError – If wait times out
- get_status(task_id: str) TaskStatus | None[source]
Get task status.
- Parameters:
task_id – Task ID
- Returns:
TaskStatus or None if not found
- get_task(task_id: str) Task | None[source]
Get task object.
- Parameters:
task_id – Task ID
- Returns:
Task object or None
- class uf.Theme(name: str, colors: dict[str, str]=<factory>, fonts: dict[str, str]=<factory>, css: str = '')[source]
UI theme configuration.
- name
Theme name
- Type:
str
- colors
Color palette
- Type:
dict[str, str]
- fonts
Font configuration
- Type:
dict[str, str]
- css
Additional custom CSS
- Type:
str
- class uf.ThemeConfig(default_theme: str = 'light', allow_toggle: bool = True, available_themes: list[str] | None = None, custom_theme: Theme | None = None)[source]
Configuration for theme system.
Example
>>> config = ThemeConfig( ... default_theme='dark', ... allow_toggle=True, ... available_themes=['light', 'dark', 'ocean'] ... )
- class uf.UfApp(funcs: Iterable[Callable], **mk_rjsf_app_kwargs)[source]
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: str, **kwargs) Any[source]
Call a function directly by name.
- Parameters:
func_name – Name of the function to call
**kwargs – Arguments to pass to the function
- Returns:
Result of the function call
- Raises:
KeyError – If function name not found
- get_spec(func_name: str) dict[source]
Get RJSF specification for a function.
- Parameters:
func_name – Name of the function
- Returns:
Dictionary with schema and uiSchema
- Raises:
KeyError – If function name not found
- class uf.UfAppTester(app)[source]
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: dict, error_type: str)[source]
Assert that error type matches.
- Parameters:
result – Result dictionary
error_type – Expected error type name
- Raises:
AssertionError – If error types don’t match
- assert_failure(result: dict, message: str = '')[source]
Assert that a result indicates failure.
- Parameters:
result – Result dictionary
message – Optional custom error message
- Raises:
AssertionError – If result is successful
- assert_result_equals(result: dict, expected: Any)[source]
Assert that result value equals expected.
- Parameters:
result – Result dictionary
expected – Expected value
- Raises:
AssertionError – If values don’t match
- class uf.UfTestClient(app)[source]
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: str, params: dict, *, expect_success: bool = True) dict[source]
Call a function with the given parameters.
- Parameters:
func_name – Name of the function to call
params – Dictionary of parameters
expect_success – Whether to expect success (raises on failure)
- Returns:
Result dictionary with ‘success’ and ‘result’ or ‘error’
- Raises:
AssertionError – If expect_success=True and call fails
- get_spec(func_name: str) dict[source]
Get RJSF specification for a function.
- Parameters:
func_name – Name of the function
- Returns:
Function specification dict
- Raises:
KeyError – If function not found
- class uf.User(username: str, password_hash: str, roles: list[str] = <factory>, permissions: list[str] = <factory>, metadata: dict = <factory>, created_at: datetime = <factory>, is_active: bool = True)[source]
User account information.
- username
Unique username
- Type:
str
- password_hash
Hashed password
- Type:
str
- roles
List of role names
- Type:
list[str]
- permissions
List of permission strings
- Type:
list[str]
- metadata
Additional user metadata
- Type:
dict
- created_at
When user was created
- Type:
datetime.datetime
- is_active
Whether account is active
- Type:
bool
- class uf.WebhookClient(url: str, headers: dict | None = None, timeout: float = 10.0, retry_count: int = 3)[source]
Client for sending webhooks.
Example
>>> client = WebhookClient('https://example.com/webhook') >>> client.send(WebhookEvent('success', 'my_func', {}, result=42))
- send(event: WebhookEvent, async_send: bool = True) bool[source]
Send a webhook event.
- Parameters:
event – WebhookEvent to send
async_send – Whether to send asynchronously
- Returns:
True if sent successfully (for sync sends)
- class uf.WebhookEvent(event_type: str, func_name: str, params: dict, result: Any = None, error: str | None = None)[source]
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
- class uf.WebhookManager[source]
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: str, events: list[str] | None = None, headers: dict | None = None, condition: Callable | None = None) None[source]
Add a webhook.
- Parameters:
url – Webhook URL
events – List of event types to trigger on (None = all)
headers – Optional HTTP headers
condition – Optional callable to filter events
- list_webhooks() list[dict][source]
List all registered webhooks.
- Returns:
List of webhook configurations
- remove_webhook(url: str) bool[source]
Remove a webhook by URL.
- Parameters:
url – Webhook URL
- Returns:
True if removed
- trigger(event_type: str, func_name: str, params: dict, result: Any = None, error: str | None = None) int[source]
Trigger webhooks for an event.
- Parameters:
event_type – Event type
func_name – Function name
params – Function parameters
result – Function result
error – Error message
- Returns:
Number of webhooks triggered
- uf.add_openapi_routes(app: Any, funcs: list[Callable], **spec_kwargs) None[source]
Add OpenAPI routes to an app.
Adds: - /openapi.json: OpenAPI specification - /docs: Swagger UI - /redoc: ReDoc UI
- Parameters:
app – Web application
funcs – List of functions
**spec_kwargs – Arguments for generate_openapi_spec
- uf.apply_field_configs(schema: dict, ui_schema: dict, field_configs: dict[str, RjsfFieldConfig]) tuple[dict, dict][source]
Apply field configurations to existing schemas.
- Parameters:
schema – JSON Schema to modify
ui_schema – UI Schema to modify
field_configs – Mapping of field names to configurations
- 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: dict, field_name: str)[source]
Assert that a field is required.
- Parameters:
spec – The specification
field_name – Name of the field
- Raises:
AssertionError – If field is not required
- uf.assert_field_type(spec: dict, field_name: str, expected_type: str)[source]
Assert that a field has the expected type.
- Parameters:
spec – The specification
field_name – Name of the field
expected_type – Expected JSON Schema type
- Raises:
AssertionError – If type doesn’t match
- uf.assert_has_field(spec: dict, field_name: str)[source]
Assert that a spec has a specific field.
- Parameters:
spec – The specification
field_name – Name of the field to check
- Raises:
AssertionError – If field not found
- uf.assert_valid_rjsf_spec(spec: dict)[source]
Assert that a specification is valid RJSF format.
- Parameters:
spec – The specification to validate
- Raises:
AssertionError – If spec is invalid
- uf.async_to_sync(async_func: Callable) Callable[source]
Convert an async function to a synchronous wrapper.
- Parameters:
async_func – Async function to wrap
- 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: Iterable[Callable]) FunctionOrganizer[source]
Automatically group functions by their module.
- Parameters:
funcs – Functions to organize
- Returns:
FunctionOrganizer with module-based groups
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: Iterable[Callable], separator: str = '_') FunctionOrganizer[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 – Functions to organize
separator – Separator character (default: “_”)
- 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: Iterable[Callable], tag_attr: str = '__uf_group__') FunctionOrganizer[source]
Automatically group functions by a tag attribute.
Functions can be tagged with a group name using an attribute.
- Parameters:
funcs – Functions to organize
tag_attr – Name of the attribute to use for grouping
- 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: str = 'default', task_queue: TaskQueue | None = None)[source]
Decorator to run function in background.
- Parameters:
queue_name – Name of the queue to use
task_queue – 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: int = 3600, backend: CacheBackend | None = None, key_func: Callable | None = None)[source]
Decorator to cache function results.
- Parameters:
ttl – Time to live in seconds
backend – Cache backend (uses global MemoryCache if None)
key_func – 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: str | None = None)[source]
Decorator to mark a function as deprecated.
- Parameters:
message – Optional deprecation message
- Returns:
Decorator function
Example
>>> @deprecated("Use new_function instead") ... def old_function(): ... pass
- uf.dict_to_pydantic(data: dict, model_class) Any[source]
Convert dictionary to Pydantic model instance.
- Parameters:
data – Dictionary of data
model_class – Pydantic model class
- Returns:
Pydantic model instance
- Raises:
ValidationError – If data doesn’t match model schema
- uf.enable_history(func: Callable, max_size: int = 100) Callable[source]
Decorator to enable call history for a function.
- Parameters:
max_size – Maximum history size
- Returns:
Decorator function
Example
>>> @enable_history ... def my_function(x: int): ... return x * 2
- uf.field_config(**field_configs: RjsfFieldConfig)[source]
Decorator to configure specific fields of a function.
- Parameters:
**field_configs – 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: Callable) bool[source]
Check if a function uses Pydantic models in its signature.
- Parameters:
func – Function to check
- Returns:
True if any parameter is a Pydantic model
- uf.generate_openapi_spec(funcs: list[Callable], title: str = 'API', version: str = '1.0.0', description: str = '', servers: list[dict] | None = None) dict[source]
Generate OpenAPI 3.0 specification.
- Parameters:
funcs – List of functions
title – API title
version – API version
description – API description
servers – Optional list of server configs
- Returns:
OpenAPI specification dictionary
- uf.get_example(func: Callable) tuple[tuple, dict] | None[source]
Get example arguments from a function.
- Parameters:
func – Function to get example from
- 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: str) RjsfFieldConfig[source]
Get a predefined field configuration by name.
- Parameters:
config_name – Name of the configuration (e.g., ‘email’, ‘multiline_text’)
- 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: Callable) dict[str, RjsfFieldConfig][source]
Get field configurations from a function.
- Parameters:
func – Function to get field configs from
- 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: Callable) list[FieldDependency][source]
Get field dependencies from a function.
- Parameters:
func – Function to get dependencies from
- Returns:
List of FieldDependency objects
- uf.get_global_auth_backend() AuthBackend | None[source]
Get the global authentication backend.
- uf.get_global_cache_backend() CacheBackend | None[source]
Get the global cache backend.
- uf.get_global_history_manager() HistoryManager[source]
Get the global history manager.
- Returns:
Global HistoryManager instance
- uf.get_global_registry() InputTransformRegistry[source]
Get the global transformation registry.
- Returns:
The global InputTransformRegistry instance
- uf.get_global_renderer_registry() ResultRendererRegistry[source]
Get the global renderer registry.
- Returns:
The global ResultRendererRegistry instance
- uf.get_global_task_queue(name: str = 'default') TaskQueue | None[source]
Get a global task queue by name.
- Parameters:
name – Queue name
- Returns:
TaskQueue or None
- uf.get_global_webhook_manager() WebhookManager | None[source]
Get the global webhook manager.
- uf.get_group(func: Callable) str | None[source]
Get group name from a function.
- Parameters:
func – Function to get group from
- 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: Callable) str | None[source]
Get the result renderer type for a function.
- Parameters:
func – Function to check
- Returns:
Renderer type string or None
- uf.get_theme(name: str) Theme | None[source]
Get a built-in theme by name.
- Parameters:
name – Theme name
- Returns:
Theme object or None
Example
>>> theme = get_theme('dark') >>> css = theme.to_css()
- uf.get_ui_config(func: Callable) dict | None[source]
Get UI configuration from a function.
- Parameters:
func – Function to get config from
- Returns:
Configuration dictionary or None if not configured
Example
>>> config = get_ui_config(my_function) >>> if config: ... print(config['title'])
- uf.group(group_name: str)[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 – Name of the group
- Returns:
Decorator function
Example
>>> @group("Admin") ... def delete_user(user_id: int): ... pass
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 – Function to hide
- Returns:
Decorated function
Example
>>> @hidden ... def internal_function(): ... pass
- uf.is_async_function(func: Callable) bool[source]
Check if a function is async.
- Parameters:
func – Function to check
- Returns:
True if function is async
Example
>>> async def my_async_func(): ... pass >>> is_async_function(my_async_func) True
Check if a function is hidden from UI.
- Parameters:
func – Function to check
- Returns:
True if function is hidden, False otherwise
Example
>>> if not is_hidden(my_function): ... # Show in UI ... pass
- uf.is_pydantic_model(obj: Any) bool[source]
Check if an object is a Pydantic model.
- Parameters:
obj – Object to check
- Returns:
True if object is a Pydantic BaseModel
- uf.make_sync_compatible(funcs: list[Callable]) list[Callable][source]
Convert any async functions in the list to sync wrappers.
- Parameters:
funcs – List of functions (may include async)
- 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: Iterable[FunctionGroup], **mk_rjsf_app_kwargs)[source]
Create a uf app with grouped function navigation.
- Parameters:
groups – 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: Iterable[Callable], *, config: Any | None = None, input_trans: Callable | None = None, output_trans: Callable | None = None, rjsf_config: dict | None = None, ui_schema_factory: Callable | None = None, param_to_prop_type: Callable | None = None, page_title: str = 'Function Interface', function_display_names: Mapping | None = None, custom_css: str | None = None, rjsf_theme: str = 'default', add_ui: bool = 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 – Iterable of callable functions to expose via web UI
config – Optional qh.AppConfig for HTTP service configuration
input_trans – Optional input transformation function for qh
output_trans – Optional output transformation function for qh
rjsf_config – Optional configuration dict for RJSF generation
ui_schema_factory – Optional callable to customize UI schema
param_to_prop_type – Optional callable to map parameters to types
page_title – Title for the web interface
function_display_names – Optional mapping to override function names
custom_css – Optional custom CSS for the web interface
rjsf_theme – RJSF theme to use (‘default’, ‘material-ui’, etc.)
add_ui – 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) dict[source]
Convert a Pydantic model to JSON Schema.
- Parameters:
model_class – Pydantic model class
- 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: Any) dict[source]
Convert Pydantic model instance to dictionary.
- Parameters:
obj – Pydantic model instance
- Returns:
Dictionary representation
- uf.rate_limit(calls: int, period: int)[source]
Decorator to mark a function with rate limiting metadata.
This is metadata-only; actual rate limiting must be implemented separately.
- Parameters:
calls – Number of calls allowed
period – 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: ResultRenderer, priority: int = 0) None[source]
Register a renderer in the global registry.
- Parameters:
renderer – ResultRenderer instance
priority – Higher priority renderers are tried first
- uf.register_type(*args, **kwargs)[source]
Register a type in the global registry.
This is a convenience function that uses the global registry. See InputTransformRegistry.register_type for full documentation.
- uf.render_result(result: Any, renderer_type: str | None = None) dict[source]
Render a result using the global registry.
- Parameters:
result – The function result to render
renderer_type – Optional specific renderer type to use
- Returns:
Rendered result dictionary
- uf.require_auth(backend: AuthBackend, roles: list[str] | None = None, permissions: list[str] | None = None)[source]
Decorator to require authentication for a function.
- Parameters:
backend – Authentication backend
roles – Required roles (any)
permissions – 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: list[str] | None = None, permissions: list[str] | None = 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 – List of required roles
permissions – 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: str)[source]
Decorator to specify result renderer for a function.
- Parameters:
renderer_type – 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: int = 3, delay: float = 1.0)[source]
Decorator to add retry logic to async functions.
- Parameters:
max_attempts – Maximum number of attempts
delay – 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: Callable, test_inputs: dict, *, expected_output: Any = None, expected_exception: type | None = None) bool[source]
Test a function with form-like input.
- Parameters:
func – Function to test
test_inputs – Dictionary of test parameters
expected_output – Expected return value (if any)
expected_exception – Expected exception type (if any)
- Returns:
True if test passes
- Raises:
AssertionError – If test fails
Example
>>> def add(x: int, y: int) -> int: ... return x + y >>> test_ui_function(add, {'x': 10, 'y': 20}, expected_output=30) True
- uf.timeout_async(seconds: float)[source]
Decorator to add timeout to async functions.
- Parameters:
seconds – Timeout in seconds
- Returns:
Decorator function
Example
>>> @timeout_async(5.0) ... async def slow_operation(): ... await asyncio.sleep(10) # Will timeout after 5s
- uf.ui_config(*, title: str | None = None, description: str | None = None, group: str | None = None, fields: dict[str, RjsfFieldConfig] | None = None, hidden: bool = False, icon: str | None = None, order: int | None = None)[source]
Decorator to add UI configuration to functions.
This decorator attaches metadata to functions that uf can use to customize the generated UI.
- Parameters:
title – Custom title for the function in UI
description – Custom description (overrides docstring)
group – Group name for organization
fields – Dictionary mapping parameter names to RjsfFieldConfig
hidden – Whether to hide this function from UI
icon – Icon identifier for the function
order – Display order within group
- Returns:
Decorator function
Example
>>> @ui_config( ... title="User Registration", ... group="Admin", ... fields={'email': RjsfFieldConfig(format='email')} ... ) ... def register_user(email: str, name: str): ... pass
- uf.webhook(on: list[str] | None = None, url: str | None = None, manager: WebhookManager | None = None)[source]
Decorator to add webhooks to a function.
- Parameters:
on – List of events to trigger on (‘success’, ‘failure’, ‘start’)
url – Optional webhook URL
manager – Optional WebhookManager instance
- Returns:
Decorator function
Example
>>> @webhook(on=['success', 'failure']) ... def process_order(order_id: int): ... # Process order ... return {'status': 'processed'}
- uf.with_dependencies(*dependencies: FieldDependency)[source]
Decorator to attach field dependencies to a function.
- Parameters:
*dependencies – FieldDependency objects
- Returns:
Decorator function
Example
>>> @with_dependencies( ... FieldDependency('reason', 'other_reason', ... lambda v: v == 'other', ... DependencyAction.SHOW) ... ) ... def submit_feedback(reason: str, other_reason: str = ''): ... pass
- uf.with_example(*example_args, **example_kwargs)[source]
Decorator to attach example arguments to a function.
This can be used to provide example/test data that appears in the UI.
- Parameters:
*example_args – Example positional arguments
**example_kwargs – Example keyword arguments
- Returns:
Decorator function
Example
>>> @with_example(x=10, y=20) ... def add(x: int, y: int) -> int: ... return x + y
- uf.wrap_pydantic_function(func: Callable) Callable[source]
Wrap a function that uses Pydantic models.
The wrapper converts dict inputs to Pydantic models before calling the function, and converts Pydantic outputs back to dicts.
- Parameters:
func – Function to wrap
- Returns:
Wrapped function
Example
>>> from pydantic import BaseModel >>> class User(BaseModel): ... name: str >>> def create_user(user: User) -> User: ... return user >>> wrapped = wrap_pydantic_function(create_user) >>> result = wrapped({'name': 'Alice'}) # Pass dict, not User