Distributed Redis State Model A lightweight Pydantic-based base class for keeping your model state in sync across multiple Python services via Redis. Each instance is identified by a unique id_ and will automatically read/write its fields to Redis whenever they change. Features Pydantic model: full data‐validation, parsing and type hints. Automatic Redis sync: reads from Redis on init and writes back on assignment. Portable: move instances between services (same class name + id_) and state stays in sync. Minimal dependencies: just pydantic and redis-py. Installation bash pip install pydantic-state-manager Quickstart ```python from src.base_redis_state_model import BaseRedisStateModel class DistributedTaskState(BaseRedisStateModel): """ Tracks the lifecycle state of a distributed task. """ current_state: str = "scheduled" Create or load the state for “task_1” task_state = DistributedTaskState(id_="task_1") Update the state; this writes immediately to Redis task_state.current_state = "running" In another service/process (with same code): state = DistributedTaskState(id_="task_1") print(state.current_state) # → "running" task_state.current_state = "completed" ``` How It Works Initialization On __init__, the model attempts to fetch any existing JSON blob from Redis at key "{ClassName}:{id_}". If found, it populates the model fields from that JSON; otherwise it uses the default values you defined in your subclass. Attribute Assignment Whenever you set any field on your model, the base class overrides __setattr__ to: Validate the new value (via Pydantic) Write it back to Redis under the same key Cross-Service Sync As long as each service imports the same subclass name and gives it the same id_, any assignment in one process will be immediately visible to others when they next access the property (or reinstantiate the object). API Reference class BaseRedisStateModel(pydantic.BaseModel) | Parameter | Type | Required | Description | | --------- | ----- | -------- | ------------------------------------------ | | id_ | str | Yes | Unique identifier for this model instance. | Configuration Under the hood, the Redis connection is managed via environment variables (or defaults): STATE_MANAGER_REDIS_URL (default: redis://localhost:6379/0) You can override these at import time: python import os os.environ["STATE_MANAGER_REDIS_URL"] = "redis-prod.mycompany.internal" License MIT © 2025 Ruslan Schalkenbajew
pypi package. Binary
Latest version: 0.0.6 Released: 2025-05-12
Pydantic Typer Typer extension to enable pydantic support [!WARNING] This package is still in early development and some things might not work as expected, or change between versions. Table of Contents Installation Usage License Installation console pip install pydantic-typer [!NOTE] pydantic-typer comes with pydantic and typer as dependencies, so you don't need to install anything else. Usage For general typer usage, please refer to the typer documentation. All the code blocks below can be copied and used directly (they are tested Python files). To run any of the examples, copy the code to a file main.py, and run it: console python main.py Basic Usage :technologist: Simply use pydantic_typer.run instead of typer.run to enable pydantic support ```python from typing import Annotated import pydantic import typer import pydantic_typer class User(pydantic.BaseModel): id: Annotated[int, pydantic.Field(description="The id of the user.")] name: Annotated[str, pydantic.Field(description="The name of the user.")] = "Jane Doe" def main(num: int, user: User): typer.echo(f"{num} {type(num)}") typer.echo(f"{user} {type(user)}") if name == "main": pydantic_typer.run(main) ``` :t-rex: Non-Annotated Version ```python import pydantic import typer import pydantic_typer class User(pydantic.BaseModel): id: int = pydantic.Field(description="The id of the user.") name: str = pydantic.Field("Jane Doe", description="The name of the user.") def main(num: int, user: User): typer.echo(f"{num} {type(num)}") typer.echo(f"{user} {type(user)}") if name == "main": pydantic_typer.run(main) ``` :computer: Usage ```console $ # Run the basic example: $ python main.py Usage: main.py [OPTIONS] NUM Try 'main.py --help' for help. ╭─ Error ────────────────────────────────────────────────────────────╮ │ Missing argument 'NUM'. │ ╰────────────────────────────────────────────────────────────────────╯ $ # We're missing a required argument, try using --help as suggested: $ python main.py --help Usage: main.py [OPTIONS] NUM ╭─ Arguments ────────────────────────────────────────────────────────╮ │ * num INTEGER [default: None] [required] │ ╰────────────────────────────────────────────────────────────────────╯ ╭─ Options ──────────────────────────────────────────────────────────╮ │ * --user.id INTEGER The id of the user. [default: None] │ │ [required] │ │ --user.name TEXT The name of the user. │ │ [default: Jane Doe] │ │ --help Show this message and exit. │ ╰──────────────────────────────────────────────────────────────────── $ # Notice the help text for user.id and user.name are inferred from the pydantic.Field. $ # user.id is reqired, because we don't provide a default value for the field. $ # Now run the example with the required arguments: $ python main.py 1 --user.id 1 1 id=1 name='Jane Doe' $ # It worked! You can also experiment with an invalid user.id: $ python main.py 1 --user.id some-string Usage: example_001_basic.py [OPTIONS] NUM Try 'example_001_basic.py --help' for help. ╭─ Error ─────────────────────────────────────────────────────────────╮ │ Invalid value for '--user.id': 'some-string' is not a valid integer.│ ╰─────────────────────────────────────────────────────────────────────╯ ``` Usage with nested models :technologist: pydantic_typer.run also works with nested pydantic models ```python from future import annotations from typing import Optional import pydantic import typer import pydantic_typer class Pet(pydantic.BaseModel): name: str species: str class Person(pydantic.BaseModel): name: str age: Optional[float] = None # noqa: UP007 For Python versions >=3.10, prefer float | None pet: Pet def main(person: Person): typer.echo(f"{person} {type(person)}") if name == "main": pydantic_typer.run(main) ``` :computer: Usage console $ # Run the nested models example with the required options: $ python main.py --person.name "Patrick" --person.pet.name "Snoopy" --person.pet.species "Dog" name='Patrick' age=None pet=Pet(name='Snoopy', species='Dog') Use pydantic models with typer.Argument :technologist: You can annotate the parameters with typer.Argument to make all model fields CLI arguments ```python from future import annotations import pydantic import typer from typing_extensions import Annotated import pydantic_typer class User(pydantic.BaseModel): id: int name: str def main(num: Annotated[int, typer.Option()], user: Annotated[User, typer.Argument()]): typer.echo(f"{num} {type(num)}") typer.echo(f"{user} {type(user)}") if name == "main": pydantic_typer.run(main) ``` :computer: Usage ```console $ # Run the example $ python main.py Usage: main.py [OPTIONS] _PYDANTIC_USER_ID _PYDANTIC_USER_NAME Try 'main.py --help' for help. ╭─ Error ─────────────────────────────────────────────────────────────╮ │ Missing argument '_PYDANTIC_USER_ID'. │ ╰─────────────────────────────────────────────────────────────────────╯ $ #Notice how _PYDANTIC_USER_ID and _PYDANTIC_USER_NAME are now cli arguments instead of options. $ # Supply the arguments in the right order: python main.py 1 Patrick --num 1 1 id=1 name='Patrick' ``` :bulb: You can also override annotations directly on the pydantic model fields: ```python from future import annotations import pydantic import typer from typing_extensions import Annotated import pydantic_typer class User(pydantic.BaseModel): id: Annotated[int, typer.Argument(metavar="THE_ID")] name: Annotated[str, typer.Option()] def main(num: Annotated[int, typer.Option()], user: Annotated[User, typer.Argument()]): typer.echo(f"{num} {type(num)}") typer.echo(f"{user} {type(user)}") if name == "main": pydantic_typer.run(main) ``` Here, User is a typer.Argument, but we manually override the fields again: We override the metavar of to User.id be THE_ID And User.name to be a typer.Option Use pydantic models in multiple commands :technologist: For larger typer apps, you can use pydantic_typer.Typer instead of annotating each command function individually to enable pydantic models on all commands ```python from future import annotations import pydantic import typer from typing_extensions import Annotated from pydantic_typer import Typer app = Typer() class User(pydantic.BaseModel): id: int name: Annotated[str, typer.Option()] = "John" @app.command() def hi(user: User): typer.echo(f"Hi {user}") @app.command() def bye(user: User): typer.echo(f"Bye {user}") if name == "main": app() ``` Use pydantic types :technologist: You can also annotate arguments with pydantic types and they will be validated ```python import click import typer from pydantic import HttpUrl, conint import pydantic_typer EvenInt = conint(multiple_of=2) def main(num: EvenInt, url: HttpUrl, ctx: click.Context): # type: ignore typer.echo(f"{num} {type(num)}") typer.echo(f"{url} {type(url)}") if name == "main": pydantic_typer.run(main) ``` :technologist: Pydantic types also work in lists and tuples ```python from typing import List import typer from pydantic import AnyHttpUrl import pydantic_typer def main(urls: List[AnyHttpUrl] = typer.Option([], "--url")): typer.echo(f"urls: {urls}") if name == "main": pydantic_typer.run(main) ``` Use Union types :technologist: Thanks to pydantic.TypeAdapter, which we use internally, we also support Union types ```python import typer import pydantic_typer def main(value: bool | int | float | str = 1): typer.echo(f"{value} {type(value)}") if name == "main": pydantic_typer.run(main) ``` :computer: Usage ```console $ # Run the example using a boolean $ python main.py --value True True $ # Run the example using an integer $ python main.py --value 2 2 $ # Run the example using a float $ python main.py --value 2.1 2.1 $ # Run the example using a string $ python main.py --value "Hello World" Hello World $ # Before, we intentionally used 2, when testing the integer $ # Check what happens if you pass 1 $ python main.py --value 1 True $ # We get back a boolean! $ # This is because Unions are generally evaluated left to right. $ # So in this case bool > int > float > str, if parsing is successful. $ # There are some exceptions, where pydantic tries to be smart, see here for details: $ # https://docs.pydantic.dev/latest/concepts/unions/#smart-mode ``` Limitations [!WARNING] pydantic-typer does not yet support sequences of pydantic models: See this issue for details [!WARNING] pydantic-typer does not yet support self-referential pydantic models. [!WARNING] pydantic-typer does not yet support lists with complex sub-types, in particular unions such as list[str|int]. License pydantic-typer is distributed under the terms of the MIT license.
Latest version: 0.0.13 Released: 2024-11-14
OSCAL Pydantic Description A simple module that contains pydantic datamodels representing the OSCAL standard. They are built from the OSCAL models published by NIST at https://github.com/usnistgov/OSCAL Several Python projects include data models, but importing a large project just to get access to the datamodel represents a significant overhead. This module simply provides the models. Installation pip install oscal-pydantic Usage To import a specific model, include it in your python file: e.g.: from oscal_pydantic import catalog Alternatively, you can import the complete OSCAL schema: from oscal_pydantic import complete After importing, you should be able to define OSCAL objects that support pydantic's rich validation rules. License This code is released under the [CC0 1.0 Universal Public Domain Dedication] (https://creativecommons.org/publicdomain/zero/1.0/).
Latest version: 2023.3.21 Released: 2023-03-21
pydantic-spark This library can convert a pydantic class to a spark schema or generate python code from a spark schema. Install bash pip install pydantic-spark Pydantic class to spark schema ```python import json from typing import Optional from pydantic_spark.base import SparkBase class TestModel(SparkBase): key1: str key2: int key2: Optional[str] schema_dict: dict = TestModel.spark_schema() print(json.dumps(schema_dict)) ``` Coerce type Pydantic-spark provides a coerce_type option that allows type coercion. When applied to a field, pydantic-spark converts the column's data type to the specified coercion type. ```python import json from pydantic import Field from pydantic_spark.base import SparkBase, CoerceType class TestModel(SparkBase): key1: str = Field(extra_json_schema={"coerce_type": CoerceType.integer}) schema_dict: dict = TestModel.spark_schema() print(json.dumps(schema_dict)) ``` Install for developers Install package Requirement: Poetry 1.* shell poetry install Run unit tests ```shell pytest coverage run -m pytest # with coverage or (depends on your local env) poetry run pytest poetry run coverage run -m pytest # with coverage ``` Run linting The linting is checked in the github workflow. To fix and review issues run this: shell black . # Auto fix all issues isort . # Auto fix all issues pflake . # Only display issues, fixing is manual
Latest version: 1.0.1 Released: 2023-11-24
Pydantic Cache Cache results of Python functions, with support for serialization of rich data types powered by Pydantic. Supports caching to disk or Redis by default, but additional caching backends can easily be added. Examples Basic usage You can use any data types which can be serialized by Pydantic, both in the function signature (cache key) and the returned values: ```python from datetime import datetime, timedelta from pydantic import BaseModel from pydantic_cache import disk_cache class MyModel(BaseModel): a: int b: datetime d: set[datetime] @disk_cache(path="~/.cache/my-function", ttl=timedelta(days=1)) def my_function(date: datetime) -> list[MyModel]: return [] # Some expensive computation ``` In the above example, subsequent calls to the function with the same argument will fetch the results from the cache on disk. Serialization and deserialization are handled based on the function's type annotations. Redis support The library includes support for caching results to/from redis. This depends on redis, which can be installed via pip install pydantic-cache[redis]. ```python from datetime import timedelta from pydantic_cache import cache from pydantic_cache.backend import RedisBackend from redis import Redis redis = Redis(...) @cache(RedisBackend(redis, ttl=timedelta(days=1))) def my_function() -> dict: return {} ``` Custom cache backends You can implement custom cache backends by sub-classing Backend: ```python from pydantic_cache import Backend, cache class MemoryBackend(Backend): def init(self) -> None: # Optional initial set-up of the backend. self._cache: dict[str, str] = {} def get(self, key: str) -> str: # Implement cache retrieval here. # Cache misses should raise a KeyError. return self._cache[key] def write(self, key: str, value: str) -> None: # Write to the cache here. self._cache[key] = value @cache(backend=MemoryBackend) def my_function() -> dict: return {} ``` [!NOTE] Cache backends only interact with serialized data, so the str types above will apply for all backends. Deferred backend resolution Some backends may rely on reading configurations or creating connections to external services, which is best avoided at import time. To support this, the cache decorator optionally accepts a callable which returns the backend, instead of the backend itself. ```python from datetime import timedelta from pathlib import Path from pydantic_cache import DiskBackend, cache from pydantic_settings import BaseSettings class Settings(BaseSettings): cache_ttl: timedelta cache_path: Path def get_cache_backend() -> DiskBackend: settings = Settings() return DiskBackend(settings.cache_path, ttl=settings.cache_ttl) @cache(backend=get_cache_backend) def my_function() -> dict: return {} ``` asyncio support Asynchronous functions are supported by default, however using a synchronous backend will naturally result in blocking calls: ```python import asyncio from pydantic_cache import DiskBackend, cache @cache(backend=DiskBackend(...)) async def my_function() -> dict: return await asyncio.sleep(0, {}) ``` To avoid blocking IO calls in the cache backend, you can implement an asynchronous backend as a subclass of AsyncBackend. See the following example using aioredis: ```python import asyncio import aioredis from pydantic_cache import AsyncBackend, cache class AioRedisBackend(AsyncBackend): def init(self, redis: aioredis.Client) -> None: self.redis = redis async def get(self, key: str) -> str: result = await self.redis.get(key) if result is None: raise KeyError(key) return result async def write(self, key: str, value: str) -> None: await self.redis.set(key, value) @cache(backend=AioRedisBackend(...)) async def my_function() -> dict: return await asyncio.sleep(0, {}) ``` Installation This project is not currently packaged and so must be installed manually. Clone the project with the following command: git clone https://github.com/jacksmith15/pydantic-cache.git Development Install dependencies: shell pyenv shell 3.10.x pre-commit install # Configure commit hooks poetry install # Install Python dependencies Run tests: shell poetry run inv verify License This project is distributed under the MIT license.
Latest version: 0.1.0 Released: 2024-02-11
sarif-pydantic An implementation of the SARIF (Static Analysis Results Interchange Format) format using Pydantic. Overview This library provides Pydantic models for working with the SARIF specification (version 2.1.0). It enables Python developers to: Create, validate, and manipulate SARIF data Parse existing SARIF files into typed Python objects Export SARIF data to JSON with proper validation Installation bash pip install sarif-pydantic Usage Creating a SARIF Log ```python from sarif_pydantic import ( ArtifactLocation, Invocation, Level, Location, Message, PhysicalLocation, Region, Result, Run, Sarif, Tool, ToolDriver ) Create a tool driver tool_driver = ToolDriver( name="Example Analyzer", version="1.0.0", ) Create a tool with the driver tool = Tool(driver=tool_driver) Create a physical location physical_location = PhysicalLocation( artifact_location=ArtifactLocation( uri="src/example.py", ), region=Region( start_line=42, start_column=5, end_line=42, end_column=32, ), ) Create a result result = Result( rule_id="EX001", level=Level.WARNING, message=Message( text="Example warning message", ), locations=[Location( physical_location=physical_location, )], ) Create a SARIF log sarif_log = Sarif( version="2.1.0", runs=[Run( tool=tool, invocations=[Invocation( execution_successful=True, )], results=[result], )], ) Export to JSON sarif_json = sarif_log.model_dump_json(indent=2, exclude_none=True) print(sarif_json) ``` Loading a SARIF Log from JSON ```python import json from sarif_pydantic import Sarif Load from a file with open("example.sarif", "r") as f: sarif_data = json.load(f) Parse into a Sarif object sarif_log = Sarif.model_validate(sarif_data) Access data via typed objects for run in sarif_log.runs: for result in run.results or []: print(f"Rule: {result.rule_id}, Level: {result.level}") print(f"Message: {result.message.text}") ``` SARIF Specification This implementation follows the SARIF 2.1.0 specification. License [LICENSE]
pypi package. Binary
Latest version: 0.5.3 Released: 2025-04-11
pydantic-kedro Advanced serialization for Pydantic models via Kedro and fsspec. This package implements custom Kedro "datasets" for both "pure" and "arbitrary" Pydantic models. You can also use it stand-alone, using Kedro just for serializing other object types. Please see the documentation for a tutorial and more examples. Usage with Kedro You can use the [PydanticAutoDataset][pydantic_kedro.PydanticAutoDataset] or any other dataset from pydantic-kedro within your Kedro catalog to save your Pydantic models: ```yaml conf/base/catalog.yml my_pydantic_model: type: pydantic_kedro.PydanticAutoDataset filepath: folder/my_model ``` Direct Dataset Usage This example works for "pure", JSON-safe Pydantic models via PydanticJsonDataset: ```python from pydantic import BaseModel from pydantic.v1 import BaseModel # Pydantic V2 from pydantic_kedro import PydanticJsonDataset class MyPureModel(BaseModel): """Your custom Pydantic model with JSON-safe fields.""" x: int y: str obj = MyPureModel(x=1, y="why?") Create an in-memory (temporary) file via fsspec and save it ds = PydanticJsonDataset("memory://temporary-file.json") ds.save(obj) We can re-load it from the same file read_obj = ds.load() assert read_obj.x == 1 ``` Standalone Usage You can also use pydantic-kedro as a generic saving and loading engine for Pydantic models: ```python from tempfile import TemporaryDirectory from pydantic.v1 import BaseModel from pydantic_kedro import load_model, save_model class MyModel(BaseModel): """My custom model.""" name: str We can use any fsspec URL, so we'll make a temporary folder with TemporaryDirectory() as tmpdir: save_model(MyModel(name="foo"), f"{tmpdir}/my_model") obj = load_model(f"{tmpdir}/my_model") assert obj.name == "foo" ```
Latest version: 0.8.0 Released: 2024-04-01
Pydantic forms A Python package that lets you add smart forms to FastAPI and Flask. Forms will respond with a JSON scheme that contains all info needed in a React frontend with uniforms to render the forms and handle all validation tasks. Forms can also consist out of a wizard, so you can create complex form flows consisting out of multiple consecutive forms. The forms and the validation logic are defined by using Pydantic models. Documentation regarding the usage of Forms can be found here Installation (Development standalone) Install the project and its dependencies to develop on the code. Step 1 - install flit: shell python3 -m venv venv source venv/bin/activate pip install flit Step 2 - install the development code: shell flit install --deps develop --symlink --python venv/bin/python !!! danger Make sure to use the flit binary that is installed in your environment. You can check the correct path by running shell which flit To be sure that the packages will be installed against the correct venv you can also prepend the python interpreter that you want to use: shell flit install --deps develop --symlink --python venv/bin/python Running tests Run the unit-test suite to verify a correct setup. Step 2 - Run tests shell pytest tests/unit_tests or with xdist: shell pytest -n auto tests/unit_tests If you do not encounter any failures in the test, you should be able to develop features in the pydantic-forms. Installation (Development symlinked into project that use pydantic-forms) If you are working on a project that already uses the pydantic-forms and you want to test your new form features against it, you can use some flit magic to symlink the dev version of the forms to your project. It will automatically replace the pypi dep with a symlink to the development version of the core and update/downgrade all required packages in your own project. Step 1 - install flit: shell python - m venv venv source venv/bin/activate pip install flit Step 2 - symlink pydantic-forms to your own project shell flit install --deps develop --symlink --python /path/to/a/project/venv/bin/python Increasing the version number for a (pre) release. When your PR is accepted you will get a version number. You can do the necessary change with a clean, e.g. every change committed, branch: shell bumpversion patch --new-version 0.0.1 Note: specifying it like this, instead of relying on bumpversion itself to increase the version, allows you to set a "RC1" version if needed. Debugging Form behaviour If you want/need the traceback of pydantic in a Form response you can add an env variable: LOG_LEVEL_PYDANTIC_FORMS=DEBUG This will add the traceback to the JSONResponse. If the loglevel is set to DEBUG the library will also add the traceback to the logger.
Latest version: 2.0.0 Released: 2025-04-09
extra-pydantic Better and more consistent pydantic support for (parametrized) generics, custom builtin-like classes, protocols, dataclasses.
Latest version: 0.1.0 Released: 2022-09-10