Python Data Structures for Humans - a library for data validation and transformation using structured models
—
Utility classes and functions supporting Schematics functionality. These components provide role-based field filtering, validation helpers, and other supporting functionality for advanced model usage.
Role-based field filtering system for controlling data access and export.
class Role(Set):
"""
Field filtering system using set-based operations.
A Role defines which fields should be included or excluded from
operations like export, serialization, or validation based on
user permissions or context.
"""
def __init__(self, function, fields):
"""
Initialize role with filtering function and field set.
Args:
function (callable): Filter function taking field name, returns bool
fields (iterable): Collection of field names this role affects
"""
def __contains__(self, field_name):
"""
Check if field is included in this role.
Args:
field_name (str): Field name to check
Returns:
bool: True if field should be included
"""Utilities for dynamic import of classes and functions from string paths.
def import_string(import_name):
"""
Import a class or function from a string path.
Args:
import_name (str): Dotted import path (e.g., 'mymodule.MyClass')
Returns:
object: Imported class or function
Raises:
ImportStringError: If import fails
"""
class ImportStringError(Exception):
"""
Exception raised when import_string fails to import from string path.
"""Helper functions for working with collections and data structures.
def listify(value):
"""
Convert value to list format for consistent processing.
Args:
value: Value to convert (single item, list, tuple, etc.)
Returns:
list: Value as list
"""
def get_all_subclasses(cls):
"""
Get all subclasses of a class recursively.
Args:
cls (type): Class to find subclasses of
Returns:
set: All subclasses including nested subclasses
"""Thread-safe utilities for concurrent model usage.
def get_ident():
"""
Get current thread identifier for thread-safe operations.
Returns:
int: Thread identifier
"""
def setdefault(obj, attr, value, search_mro=False, overwrite_none=False):
"""
Enhanced setdefault with MRO search and None overwrite options.
Args:
obj: Object to set attribute on
attr (str): Attribute name
value: Default value to set
search_mro (bool): Whether to search method resolution order
overwrite_none (bool): Whether to overwrite None values
Returns:
Current or newly set attribute value
"""Special constant classes used throughout Schematics.
class Constant(int):
"""
Special integer constant with string representation.
Used for export levels and other enumerated constants that need
both numeric and readable string representations.
"""
def __new__(cls, value, name=None):
"""
Create constant with value and optional name.
Args:
value (int): Numeric value
name (str, optional): String representation
"""from schematics.models import Model
from schematics.types import StringType, IntType, BooleanType
from schematics.role import Role
class User(Model):
name = StringType(required=True)
email = StringType(required=True)
age = IntType()
is_admin = BooleanType(default=False)
salary = IntType()
# Define roles
public_role = Role(lambda field: field in ['name', 'age'], ['name', 'age'])
admin_role = Role(lambda field: field != 'salary', ['name', 'email', 'age', 'is_admin'])
user = User({
'name': 'John Doe',
'email': 'john@example.com',
'age': 30,
'is_admin': False,
'salary': 75000
})
# Export with role filtering
public_data = user.export(role=public_role)
# {'name': 'John Doe', 'age': 30}
admin_data = user.export(role=admin_role)
# {'name': 'John Doe', 'email': 'john@example.com', 'age': 30, 'is_admin': False}from schematics.util import import_string, listify
# Dynamic import of field types
field_class = import_string('schematics.types.StringType')
field = field_class(required=True, max_length=100)
# Convert various inputs to lists
single_value = listify('item') # ['item']
multiple_values = listify(['a', 'b']) # ['a', 'b']
tuple_values = listify(('x', 'y')) # ['x', 'y']import threading
from schematics.util import get_ident
class ThreadSafeModel(Model):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._thread_id = get_ident()
def validate(self, *args, **kwargs):
current_thread = get_ident()
if current_thread != self._thread_id:
# Handle cross-thread validation
pass
return super().validate(*args, **kwargs)from schematics.util import Constant
# Define custom export levels
BASIC_EXPORT = Constant(5, 'BASIC_EXPORT')
DETAILED_EXPORT = Constant(15, 'DETAILED_EXPORT')
FULL_EXPORT = Constant(25, 'FULL_EXPORT')
class Product(Model):
name = StringType(export_level=BASIC_EXPORT)
description = StringType(export_level=DETAILED_EXPORT)
internal_notes = StringType(export_level=FULL_EXPORT)
# Export at different levels
product = Product({
'name': 'Widget',
'description': 'A useful widget',
'internal_notes': 'Manufactured in facility B'
})
# Only exports fields with export_level <= specified level
basic = product.export(export_level=BASIC_EXPORT)
# {'name': 'Widget'}
detailed = product.export(export_level=DETAILED_EXPORT)
# {'name': 'Widget', 'description': 'A useful widget'}
full = product.export(export_level=FULL_EXPORT)
# {'name': 'Widget', 'description': 'A useful widget', 'internal_notes': 'Manufactured in facility B'}Install with Tessl CLI
npx tessl i tessl/pypi-schematics