Pydantic

Integration for instrumenting Pydantic models.

PluginSettings

Bases: TypedDict

A typed dict for the Pydantic plugin settings.

This is how you can use the PluginSettings with a Pydantic model:

from logfire.integrations.pydantic import PluginSettings
from pydantic import BaseModel


class Model(BaseModel, plugin_settings=PluginSettings(logfire={'record': 'all'})):
    a: int

logfire instance-attribute

logfire: LogfireSettings

Settings for the logfire integration.

LogfireSettings

Bases: TypedDict

Settings for the logfire integration.

trace_sample_rate instance-attribute

trace_sample_rate: float

The sample rate to use for tracing.

tags instance-attribute

tags: list[str]

Tags to add to the spans.

record instance-attribute

record: Literal['all', 'failure', 'metrics']

What to record.

The following values are supported:

• all: Record all validation events.
• failure: Record only validation failures.
• metrics: Record only validation metrics.

LogfirePydanticPlugin dataclass

LogfirePydanticPlugin()

Implements a new API for pydantic plugins.

Patches Pydantic to accept this new API shape.

Set the LOGFIRE_PYDANTIC_RECORD environment variable to "off" to disable the plugin, or PYDANTIC_DISABLE_PLUGINS to true to disable all Pydantic plugins.

new_schema_validator

new_schema_validator(
    schema: CoreSchema,
    schema_type: Any,
    schema_type_path: SchemaTypePath,
    schema_kind: SchemaKind,
    config: CoreConfig | None,
    plugin_settings: dict[str, Any],
) -> tuple[_ValidateWrapper, ...] | tuple[None, ...]

This method is called every time a new SchemaValidator is created.

Parameters:

Name	Type	Description	Default
schema	CoreSchema	The schema to validate against.	required
schema_type	Any	The original type which the schema was created from, e.g. the model class.	required
schema_type_path	SchemaTypePath	Path defining where schema_type was defined, or where TypeAdapter was called.	required
schema_kind	SchemaKind	The kind of schema to validate against.	required
config	CoreConfig | None	The config to use for validation.	required
plugin_settings	dict[str, Any]	The plugin settings.	required

Returns:

Type	Description
tuple[_ValidateWrapper, ...] | tuple[None, ...]	A tuple of decorator factories for each of the three validation methods - validate_python, validate_json, validate_strings or a tuple of three None if recording is off.

Source code in logfire/integrations/pydantic.py

def new_schema_validator(
    self,
    schema: CoreSchema,
    schema_type: Any,
    schema_type_path: SchemaTypePath,
    schema_kind: SchemaKind,
    config: CoreConfig | None,
    plugin_settings: dict[str, Any],
) -> tuple[_ValidateWrapper, ...] | tuple[None, ...]:
    """This method is called every time a new `SchemaValidator` is created.

    Args:
        schema: The schema to validate against.
        schema_type: The original type which the schema was created from, e.g. the model class.
        schema_type_path: Path defining where `schema_type` was defined, or where `TypeAdapter` was called.
        schema_kind: The kind of schema to validate against.
        config: The config to use for validation.
        plugin_settings: The plugin settings.

    Returns:
        A tuple of decorator factories for each of the three validation methods -
            `validate_python`, `validate_json`, `validate_strings` or a tuple of
            three `None` if recording is `off`.
    """
    # Patch a bug that occurs even if the plugin is disabled.
    _patch_PluggableSchemaValidator()

    logfire_settings = plugin_settings.get('logfire')
    if logfire_settings and 'record' in logfire_settings:
        record = logfire_settings['record']
    else:
        record = get_pydantic_plugin_config().record

    if record == 'off':
        return None, None, None

    if _include_model(schema_type_path):
        _patch_build_wrapper()
        return (
            _ValidateWrapper('validate_python', schema, config, plugin_settings, schema_type_path, record),
            _ValidateWrapper('validate_json', schema, config, plugin_settings, schema_type_path, record),
            _ValidateWrapper('validate_strings', schema, config, plugin_settings, schema_type_path, record),
        )

    return None, None, None

get_schema_name

get_schema_name(schema: CoreSchema) -> str

Find the best name to use for a schema.

The follow rules are used: * If the schema represents a model or dataclass, use the name of the class. * If the root schema is a wrap/before/after validator, look at its schema property. * Otherwise use the schema's type property.

Parameters:

Name	Type	Description	Default
schema	CoreSchema	The schema to get the name for.	required

Returns:

Type	Description
str	The name of the schema.

Source code in logfire/integrations/pydantic.py

def get_schema_name(schema: CoreSchema) -> str:
    """Find the best name to use for a schema.

    The follow rules are used:
    * If the schema represents a model or dataclass, use the name of the class.
    * If the root schema is a wrap/before/after validator, look at its `schema` property.
    * Otherwise use the schema's `type` property.

    Args:
        schema: The schema to get the name for.

    Returns:
        The name of the schema.
    """
    if schema['type'] in {'model', 'dataclass'}:
        return schema['cls'].__name__  # type: ignore
    elif schema['type'] in {'function-after', 'function-before', 'function-wrap'}:
        return get_schema_name(schema['schema'])  # type: ignore
    elif schema['type'] == 'definitions':
        inner_schema = schema['schema']
        if inner_schema['type'] == 'definition-ref':
            schema_ref: str = inner_schema['schema_ref']  # type: ignore
            [schema_definition] = [
                definition
                for definition in schema['definitions']
                if definition['ref'] == schema_ref  # type: ignore
            ]
            return get_schema_name(schema_definition)
        else:
            return get_schema_name(inner_schema)
    else:
        return schema['type']

get_pydantic_plugin_config

get_pydantic_plugin_config() -> PydanticPlugin

Get the Pydantic plugin config.

Source code in logfire/integrations/pydantic.py

def get_pydantic_plugin_config() -> PydanticPlugin:
    """Get the Pydantic plugin config."""
    if _pydantic_plugin_config_value is not None:
        return _pydantic_plugin_config_value
    else:
        return GLOBAL_CONFIG.param_manager.pydantic_plugin

set_pydantic_plugin_config

set_pydantic_plugin_config(
    plugin_config: PydanticPlugin | None,
) -> None

Set the pydantic plugin config.

Source code in logfire/integrations/pydantic.py

def set_pydantic_plugin_config(plugin_config: PydanticPlugin | None) -> None:
    """Set the pydantic plugin config."""
    global _pydantic_plugin_config_value
    _pydantic_plugin_config_value = plugin_config