Skip to content

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
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
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 = _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
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
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']