Extending Dracon

You need a custom instruction tag or a custom include loader.

Custom instruction tags

Instruction tags are processed during composition. They can transform the YAML tree before construction. Built-in examples include !define, !if, !each, !assert, and !require.

The Instruction class

Subclass Instruction and implement two methods:

from dracon.instructions import Instruction, register_instruction
from dracon.composer import CompositionResult
from dracon.keypath import KeyPath

class ValidatePort(Instruction):
    """Check that a port value is in a valid range during composition."""

    @staticmethod
    def match(value):
        if value == '!validate_port':
            return ValidatePort()
        return None

    def process(self, comp_res: CompositionResult, path: KeyPath, loader):
        from dracon.instructions import unpack_mapping_key
        from dracon.diagnostics import CompositionError

        key_node, value_node, parent_node = unpack_mapping_key(
            comp_res, path, 'validate_port'
        )

        # evaluate the value to get the actual port number
        port_value = value_node.value
        if isinstance(port_value, str) and port_value.isdigit():
            port_value = int(port_value)

        if not isinstance(port_value, int) or not (1 <= port_value <= 65535):
            raise CompositionError(
                f"Invalid port: {port_value} (must be 1-65535)"
            )

        # remove the instruction key from the parent mapping
        del parent_node[str(path[-1])]

        return comp_res

Registering

register_instruction('!validate_port', ValidatePort)

The ! prefix is added automatically if you forget it, so register_instruction('validate_port', ValidatePort) also works.

Using it

server:
  !validate_port port: 8080
  host: 0.0.0.0

During composition, ValidatePort.process() runs. If the port is out of range, you get a CompositionError.

match() details

The match() method receives the tag string (e.g., '!validate_port') and returns either an Instruction instance or None. This lets you support parameterized tags:

@staticmethod
def match(value):
    import re
    m = re.match(r'!validate_range\((\d+),(\d+)\)', value)
    if m:
        return ValidateRange(int(m.group(1)), int(m.group(2)))
    return None

Deferred instructions

Set deferred = True on your class to make the instruction run after all regular instructions, during the assertion pass. This is how !assert works:

class MyCheck(Instruction):
    deferred = True  # runs after !define, !if, !each, etc.

    @staticmethod
    def match(value):
        ...

    def process(self, comp_res, path, loader):
        ...

Custom include loaders

Dracon's !include tag supports scheme-prefixed paths like file:, pkg:, env:, var:. You can add your own schemes.

Loader function signature

A loader function takes a path string and returns a tuple of (content_string, context_dict):

def my_loader(path, node=None, draconloader=None):
    """
    Args:
        path: the part after the scheme prefix (e.g., "secret/data/myapp#password"
              for "vault:secret/data/myapp#password")
        node: the IncludeNode (has .context, .optional, etc.)
        draconloader: the current DraconLoader instance

    Returns:
        tuple of (content_string, context_dict)
        - content_string: YAML string to parse, or a CompositionResult
        - context_dict: extra context variables to inject (e.g., FILE_PATH)
    """
    content = fetch_from_somewhere(path)
    return content, {'SOURCE': f'my_loader:{path}'}

Registering

Pass custom_loaders to the DraconLoader:

import dracon

loader = dracon.DraconLoader(
    custom_loaders={'vault': vault_loader}
)
config = loader.load('config.yaml')

This adds your loader alongside the built-in ones (file, pkg, env, var, raw, rawpkg, cascade).

Example: HashiCorp Vault loader

import hvac

def vault_loader(path, node=None, draconloader=None):
    """Fetch secrets from HashiCorp Vault.

    Usage in YAML: !include vault:secret/data/myapp#password
    """
    # split path and key
    if '#' in path:
        vault_path, key = path.rsplit('#', 1)
    else:
        vault_path, key = path, None

    client = hvac.Client()
    response = client.secrets.kv.v2.read_secret_version(path=vault_path)
    data = response['data']['data']

    if key:
        # return a single value as a YAML scalar
        return str(data[key]), {}
    else:
        # return the whole secret as YAML
        import yaml
        return yaml.dump(data), {}

# register it
loader = dracon.DraconLoader(
    custom_loaders={'vault': vault_loader}
)

Using it

database:
  host: db.example.com
  password: !include vault:secret/data/myapp#password

Returning a CompositionResult

For more control, your loader can return a CompositionResult directly instead of a string. This skips the YAML parsing step:

from dracon.composer import CompositionResult
from dracon.nodes import DraconScalarNode

def my_loader(path, node=None, draconloader=None):
    value = compute_value(path)
    scalar = DraconScalarNode(tag='tag:yaml.org,2002:str', value=str(value))
    return CompositionResult(root=scalar), {}

The built-in loaders all follow this same interface, so you can look at dracon/loaders/ for more examples.