### Install fluxconf Source: https://github.com/greenroom-robotics/fluxconf/blob/main/README.md Install the fluxconf package using pip or uv. ```sh pip install fluxconf ``` ```sh uv add fluxconf ``` -------------------------------- ### Install and Run Pre-commit Hooks Source: https://github.com/greenroom-robotics/fluxconf/blob/main/CONTRIBUTING.md Installs pre-commit hooks for automatic quality checks and demonstrates how to run them manually on all files. ```sh # Install pre-commit hooks uv tool install pre-commit pre-commit install ``` ```sh # Run hooks manually on all files pre-commit run --all-files ``` -------------------------------- ### Install Development Dependencies Source: https://github.com/greenroom-robotics/fluxconf/blob/main/CONTRIBUTING.md Installs the package and its development dependencies using either the 'just' command runner or 'uv'. ```sh just dev ``` ```sh uv sync --extra dev ``` -------------------------------- ### Auto-Apply Migrations with Lambda Function Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt Demonstrates reading a configuration file and automatically applying pending migrations using a lambda function for data transformation. This example simulates an old config file and shows how fluxconf updates it in place. ```python import yaml from pathlib import Path from fluxconf import ConfigIO, VersionedBaseModel class UserConfig(VersionedBaseModel): username: str = "" email: str = "" class UserConfigIO(ConfigIO[UserConfig]): file_name = "user.yml" config_type = UserConfig migrations = { "1_merge_name": lambda data: { **{k: v for k, v in data.items() if k not in ("first_name", "last_name")}, "username": f"{data.get('first_name', '')} {data.get('last_name', '')}".strip() or data.get("username", ""), }, } # Simulate an old config file on disk config_dir = Path("/tmp/myapp") config_dir.mkdir(exist_ok=True) (config_dir / "user.yml").write_text( yaml.dump({"first_name": "Ada", "last_name": "Lovelace", "email": "ada@example.com"}) ) io = UserConfigIO(config_dir) config = io.read() # Migration ran automatically: print(config.username) # "Ada Lovelace" print(config.email) # "ada@example.com" print(config.version) # 1 # File on disk is now updated (old fields replaced, version stamped) raw = io._read_raw() print(raw) # {'username': 'Ada Lovelace', 'email': 'ada@example.com', 'version': 1} ``` -------------------------------- ### Run Code Formatting Source: https://github.com/greenroom-robotics/fluxconf/blob/main/CONTRIBUTING.md Formats code using Ruff according to the project's style guidelines. Configuration is detailed in pyproject.toml. ```sh just format ``` -------------------------------- ### Python Function Migrations for User Configuration Source: https://github.com/greenroom-robotics/fluxconf/blob/main/README.md Implement complex configuration migrations using Python functions. Each function receives and returns a dictionary representing the config data. ```python from fluxconf import ConfigIO, VersionedBaseModel class UserConfig(VersionedBaseModel): full_name: str = "" email: str = "" def merge_name_fields(data: dict) -> dict: first = data.pop("first_name", "") last = data.pop("last_name", "") if first or last: data["full_name"] = f"{first} {last}".strip() return data class UserConfigIO(ConfigIO[UserConfig]): file_name = "user.yml" config_type = UserConfig migrations = { "1_merge_name": merge_name_fields, } ``` -------------------------------- ### Run Migrations with Custom Version Field Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt Demonstrates how to run database migrations using a custom field name for tracking the version. This is useful when the default 'version' field conflicts with existing schema. ```python # {'version': 3, 'database_url': 'postgres://localhost/mydb', 'timeout': 30} # Migrate to a specific target version result = run_migrations({}, migrations, target_version=2) print(result["version"]) # 2 # Error handling bad_migrations = { "1_ok": lambda data: {**data, "ok": True}, "2_fail": lambda data: (_ for _ in ()).throw(RuntimeError("disk full")), } try: run_migrations({}, bad_migrations) except MigrationError as e: print(e.last_successful_migration) # 1 print(type(e.original_error)) # # ValueError when config is newer than any known migration try: run_migrations({"version": 99}, migrations) except ValueError as e: print(e) # Stored version 99 is ahead of the latest known migration 3... # Custom version field name result = run_migrations( {"schema_version": 0, "data": "x"}, {"1_tag": lambda d: {**d, "tagged": True}}, version_field="schema_version", ) print(result["schema_version"]) # 1 ``` -------------------------------- ### Basic Configuration with ConfigIO Source: https://github.com/greenroom-robotics/fluxconf/blob/main/README.md Define a Pydantic model and use ConfigIO to read and write YAML-backed configurations. Ensure the config type and file name are set in the subclass. ```python from pydantic import BaseModel from fluxconf import ConfigIO class AppConfig(BaseModel): name: str = "my-app" debug: bool = False class AppConfigIO(ConfigIO[AppConfig]): file_name = "app.yml" config_type = AppConfig io = AppConfigIO("~/.config/my-app") io.write(AppConfig(name="my-app", debug=True)) config = io.read() # AppConfig(name='my-app', debug=True) ``` -------------------------------- ### Load Migrations from Directory Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt Scans a directory for Python or JSON files containing migration logic. Python files should define a `migrate` function or `patch` attribute, while JSON files use RFC 6902 operations. Files with underscore prefixes or missing integer prefixes are ignored. ```python # Directory layout: # myapp/migrations/ # _helpers.py ← skipped (underscore prefix) # 1_rename_host.json ← JSON Patch # 2_merge_names.py ← Python migrate function # 3_add_defaults.py ← Python patch attribute # README.md ← skipped (not .py/.json) # myapp/migrations/1_rename_host.json # [{"op": "move", "from": "/hostname", "path": "/host"}] # myapp/migrations/2_merge_names.py # def migrate(data: dict) -> dict: # first = data.pop("first_name", "") # last = data.pop("last_name", "") # data["full_name"] = f"{first} {last}".strip() # return data # myapp/migrations/3_add_defaults.py # patch = [ # {"op": "add", "path": "/port", "value": 8080}, # {"op": "add", "path": "/retries", "value": 3}, # ] from pathlib import Path from fluxconf.migration import load_migrations_from_dir, run_migrations from fluxconf import ConfigIO, VersionedBaseModel class AppConfig(VersionedBaseModel): host: str = "localhost" port: int = 8080 full_name: str = "" retries: int = 3 class AppConfigIO(ConfigIO[AppConfig]): file_name = "app.yml" config_type = AppConfig migrations_dir = Path(__file__).parent / "migrations" # Manual usage of load_migrations_from_dir migrations = load_migrations_from_dir(Path("myapp/migrations")) print(list(migrations.keys())) # ['1_rename_host', '2_merge_names', '3_add_defaults'] result = run_migrations( {"hostname": "db.internal", "first_name": "Jane", "last_name": "Doe"}, migrations, ) print(result["host"]) # "db.internal" print(result["full_name"]) # "Jane Doe" print(result["port"]) # 8080 print(result["version"]) # 3 # Combine migrations_dir with inline migrations (no key collisions allowed) class HybridConfigIO(ConfigIO[AppConfig]): file_name = "app.yml" config_type = AppConfig migrations_dir = Path("myapp/migrations") migrations = { "4_set_timeout": [{"op": "add", "path": "/timeout", "value": 60}], } ``` -------------------------------- ### Run All Code Checking Tools Source: https://github.com/greenroom-robotics/fluxconf/blob/main/CONTRIBUTING.md Executes all configured code checking tools, including linting and formatting, to ensure code quality and consistency. ```sh just check ``` -------------------------------- ### Apply Migrations to a Raw Config Dict Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt The `run_migrations` function applies a subset of migrations to a plain dictionary, from a stored version up to a target version. It raises MigrationError on failure and ValueError if the stored version is too new. ```python from fluxconf.migration import run_migrations, MigrationError migrations = { "1_add_env": lambda data: {**data, "env": "production"}, "2_rename_key": lambda data: { **{k: v for k, v in data.items() if k != "db_url"}, "database_url": data.get("db_url", ""), }, "3_set_defaults": [ {"op": "add", "path": "/timeout", "value": 30}, ], } # Migrate from scratch (no stored version) result = run_migrations({}, migrations) print(result) # {'env': 'production', 'database_url': '', 'timeout': 30, 'version': 3} # Migrate from a partially-applied state (version 1 stored, only 2 and 3 run) result = run_migrations({"version": 1, "db_url": "postgres://localhost/mydb"}, migrations) print(result) ``` -------------------------------- ### Run Linting Source: https://github.com/greenroom-robotics/fluxconf/blob/main/CONTRIBUTING.md Applies linting rules using Ruff to identify potential issues and enforce code style. Configuration is detailed in pyproject.toml. ```sh just lint ``` -------------------------------- ### Define and Use ConfigIO for App Configuration Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt Subclass ConfigIO to manage application configuration with Pydantic models. Specify file name, config type, and optionally a schema URL for editor support. Demonstrates writing and reading configuration, including handling default values and serializing to YAML. ```python from pathlib import Path from pydantic import BaseModel from fluxconf import ConfigIO class AppConfig(BaseModel): name: str = "my-app" debug: bool = False workers: int = 4 class AppConfigIO(ConfigIO[AppConfig]): file_name = "app.yml" config_type = AppConfig schema_url = "https://example.com/app-schema.json" # optional: adds yaml-language-server header io = AppConfigIO("~/.config/my-app") # Write config — only non-default values are serialised by default io.write(AppConfig(name="production", debug=False, workers=8)) # Write all fields including defaults io.write(AppConfig(name="production"), include_defaults=True) # Read back — returns a fully typed AppConfig instance config = io.read() print(config.name) # "production" print(config.workers) # 8 # Get the resolved path to the config file print(io.get_path()) # PosixPath('/home/user/.config/my-app/app.yml') # Serialise to YAML string without touching disk yaml_str = io.serialise(config) print(yaml_str) # name: production # workers: 8 ``` -------------------------------- ### JSON Patch Operations for Migration Source: https://github.com/greenroom-robotics/fluxconf/blob/main/README.md Use JSON files with an array of patch operations for simple configuration updates. Files must have an integer prefix. ```json [ {"op": "move", "from": "/hostname", "path": "/host"} ] ``` -------------------------------- ### Configuring Migrations Directory Source: https://github.com/greenroom-robotics/fluxconf/blob/main/README.md Specify a directory for migration files by setting the `migrations_dir` attribute in your `ConfigIO` subclass. This allows Fluxconf to load migrations from separate files. ```python from pathlib import Path from fluxconf import ConfigIO, VersionedBaseModel class AppConfigIO(ConfigIO[AppConfig]): file_name = "app.yml" config_type = AppConfig migrations_dir = Path(__file__).parent / "migrations" ``` -------------------------------- ### JSON Patch Migrations for Server Configuration Source: https://github.com/greenroom-robotics/fluxconf/blob/main/README.md Use JSON Patch operations to define migrations for evolving configuration schemas. Inherit from VersionedBaseModel and define migrations in a dictionary. ```python from fluxconf import ConfigIO, VersionedBaseModel class ServerConfig(VersionedBaseModel): host: str = "localhost" port: int = 8080 class ServerConfigIO(ConfigIO[ServerConfig]): file_name = "server.yml" config_type = ServerConfig migrations = { "1_rename_host": [ {"op": "move", "from": "/hostname", "path": "/host"}, ], "2_add_port": [ {"op": "add", "path": "/port", "value": 8080}, ], } ``` -------------------------------- ### Inline Migrations with JSON Patch Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt Define migrations using JSON Patch operations within a ConfigIO subclass. Supported operations include add, remove, replace, move, and copy. Migration keys follow the 'N_description' format for ordering. ```python from fluxconf import ConfigIO, VersionedBaseModel class DbConfig(VersionedBaseModel): host: str = "localhost" port: int = 5432 pool_size: int = 5 class DbConfigIO(ConfigIO[DbConfig]): file_name = "db.yml" config_type = DbConfig migrations = { # Rename field "1_rename_server_to_host": [ {"op": "move", "from": "/server", "path": "/host"}, ], # Add a new field with a default "2_add_pool_size": [ {"op": "add", "path": "/pool_size", "value": 5}, ], # Update a nested value "3_default_port": [ {"op": "replace", "path": "/port", "value": 5432}, ], # Multi-operation patch "4_restructure": [ {"op": "remove", "path": "/legacy_flag"}, {"op": "add", "path": "/ssl", "value": False}, ], } ``` -------------------------------- ### Run Tests with Pytest Source: https://github.com/greenroom-robotics/fluxconf/blob/main/CONTRIBUTING.md Executes tests using Pytest. Use the first command to run tests on your current Python version, or the second to run across all supported versions. ```sh # Run tests on your current Python version uv run --extra test pytest -v ``` ```sh # Run tests across all supported Python versions just test ``` -------------------------------- ### Inline Migrations with Python Functions Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt Use Python callables for complex migrations that JSON Patch cannot express, such as conditional logic or computed fields. Callables and JSON Patch lists can be mixed in the migrations dictionary. ```python from fluxconf import ConfigIO, VersionedBaseModel class ProfileConfig(VersionedBaseModel): display_name: str = "" role: str = "viewer" permissions: list[str] = [] def promote_admin(data: dict) -> dict: """Convert old boolean 'is_admin' flag to role/permissions.""" if data.pop("is_admin", False): data["role"] = "admin" data["permissions"] = ["read", "write", "delete"] return data def normalize_display_name(data: dict) -> dict: first = data.pop("first", "") last = data.pop("last", "") data.setdefault("display_name", f"{first} {last}".strip()) return data class ProfileConfigIO(ConfigIO[ProfileConfig]): file_name = "profile.yml" config_type = ProfileConfig migrations = { "1_merge_name": normalize_display_name, # Python function "2_promote_admin": promote_admin, # Python function "3_add_permissions": [ {"op": "add", "path": "/permissions", "value": []}, ], } ``` -------------------------------- ### Python Patch Operations for Migration Source: https://github.com/greenroom-robotics/fluxconf/blob/main/README.md Define configuration changes using a Python list of patch operations, similar to JSON patch files. This requires a `patch` attribute in the Python file. ```python patch = [ {"op": "add", "path": "/port", "value": 8080}, {"op": "add", "path": "/retries", "value": 3}, ] ``` -------------------------------- ### Implement VersionedBaseModel and Migrations Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt Use VersionedBaseModel for Pydantic models that require schema evolution tracking. Define migrations as a dictionary mapping version names to JSON Patch operations or callables. The ConfigIO automatically applies these migrations on read. ```python from fluxconf import ConfigIO, VersionedBaseModel class ServerConfig(VersionedBaseModel): host: str = "localhost" port: int = 8080 tls: bool = False class ServerConfigIO(ConfigIO[ServerConfig]): file_name = "server.yml" config_type = ServerConfig migrations = { "1_add_tls": [{"op": "add", "path": "/tls", "value": False}], "2_rename_host": [{"op": "move", "from": "/hostname", "path": "/host"}], } io = ServerConfigIO("/etc/myapp") config = io.read() # After reading, config.version == 2 (or the latest applied migration number) print(config.version) # 2 ``` -------------------------------- ### Python Migrate Function for Full Flexibility Source: https://github.com/greenroom-robotics/fluxconf/blob/main/README.md Implement custom migration logic using a `migrate` function in a Python file for complex transformations. This function receives the data dictionary and returns the modified dictionary. ```python def migrate(data: dict) -> dict: first = data.pop("first_name", "") last = data.pop("last_name", "") if first or last: data["full_name"] = f"{first} {last}".strip() return data ``` -------------------------------- ### Embed YAML Language Server Schema Header Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt Set `schema_url` on a `ConfigIO` subclass to prepend a schema header comment. This enables IDE schema validation and autocompletion for editors supporting the YAML Language Server protocol. ```python from pydantic import BaseModel from fluxconf import ConfigIO class PipelineConfig(BaseModel): stages: list[str] = [] timeout: int = 600 class PipelineConfigIO(ConfigIO[PipelineConfig]): file_name = "pipeline.yml" config_type = PipelineConfig schema_url = "https://schemas.mycompany.com/pipeline/v1.json" io = PipelineConfigIO("/tmp/ci") io.write(PipelineConfig(stages=["build", "test", "deploy"])) content = io.get_path().read_text() print(content) ``` -------------------------------- ### Write Config Model to YAML with Fluxconf Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt Serializes a Pydantic model to YAML, optionally including default fields. Parent directories are created automatically. For VersionedBaseModel, the version is bumped before writing. ```python from pydantic import BaseModel from fluxconf import ConfigIO, VersionedBaseModel class CacheConfig(VersionedBaseModel): backend: str = "redis" ttl: int = 300 max_size: int = 1000 class CacheConfigIO(ConfigIO[CacheConfig]): file_name = "cache.yml" config_type = CacheConfig always_include_fields = ["version"] # force version into output even if it equals the default migrations = { "1_initial": lambda data: data, } io = CacheConfigIO("/tmp/myapp/conf") # Only non-default fields written (plus always_include_fields) io.write(CacheConfig(backend="memcached")) # Resulting YAML: # backend: memcached # version: 1 # Include every field regardless of default value io.write(CacheConfig(backend="memcached"), include_defaults=True) # Resulting YAML: # backend: memcached # max_size: 1000 # ttl: 300 # version: 1 ``` -------------------------------- ### Handle Migration Errors with Rollback Metadata Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt Catch `MigrationError` to access `last_successful_migration` and `original_error`. This allows for implementing logging, alerting, or partial rollback strategies. ```python from fluxconf.migration import run_migrations, MigrationError def step_one(data: dict) -> dict: data["processed"] = True return data def step_two(data: dict) -> dict: raise IOError("remote config service unreachable") migrations = { "1_process": step_one, "2_sync": step_two, } try: run_migrations({"version": 0}, migrations) except MigrationError as err: print(f"Migration failed: {err}") print(f"Last successful migration version: {err.last_successful_migration}") # 1 print(f"Root cause: {type(err.original_error).__name__}: {err.original_error}") # → Root cause: IOError: remote config service unreachable # Use last_successful_migration for alerting or to determine safe fallback if err.last_successful_migration >= 1: print("Data was at least partially processed before failure.") ``` -------------------------------- ### Force Fields into Serialized Output Source: https://context7.com/greenroom-robotics/fluxconf/llms.txt Ensures specific fields are always present in the serialized YAML output, even if their values match Pydantic defaults. This is useful for fields like version strings or discriminators that downstream systems require unconditionally. ```python from pydantic import BaseModel from fluxconf import ConfigIO class ServiceConfig(BaseModel): version: str = "0.0.0" # always needed by the deploy pipeline name: str = "service" replicas: int = 1 class ServiceConfigIO(ConfigIO[ServiceConfig]): file_name = "service.yml" config_type = ServiceConfig always_include_fields = ["version"] # persisted even when value == "0.0.0" io = ServiceConfigIO("/tmp/deploy") io.write(ServiceConfig(name="api-gateway")) raw = io._read_raw() print("version" in raw) # True ← always present print("replicas" in raw) # False ← default, excluded print(raw["version"]) # "0.0.0" ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.