tessl/pypi-types-pyyaml

Type stubs for PyYAML, a full-featured YAML framework for Python

Overview

Eval results

Files

Low-Level Processing

Name: tessl/pypi-types-pyyaml
Author: tessl

Lower-level YAML processing functions for advanced use cases requiring fine-grained control over the processing pipeline. These functions provide direct access to each stage of YAML processing.

Capabilities

Scanning and Parsing

Functions that provide access to the tokenization and parsing stages of YAML processing.

def scan(stream, Loader=None):
    """
    Scan YAML stream and yield tokens.
    
    Parameters:
    - stream: YAML input as string, bytes, or file-like object
    - Loader: Loader class to use for scanning (defaults to SafeLoader)
    
    Yields:
    - Token objects representing YAML syntax elements
    
    Raises:
    - ScannerError: If invalid YAML syntax is encountered
    """

def parse(stream, Loader=None):
    """
    Parse YAML stream and yield events.
    
    Parameters:
    - stream: YAML input as string, bytes, or file-like object
    - Loader: Loader class to use for parsing (defaults to SafeLoader)
    
    Yields:
    - Event objects representing YAML structure elements
    
    Raises:
    - ParserError: If invalid YAML structure is encountered
    """

Composition

Functions that build node trees from YAML events, handling anchors and aliases.

def compose(stream, Loader=None):
    """
    Compose YAML stream into a node tree.
    
    Parameters:
    - stream: YAML input as string, bytes, or file-like object
    - Loader: Loader class to use for composition (defaults to SafeLoader)
    
    Returns:
    - Root Node object representing the document structure
    
    Raises:
    - ComposerError: If node composition fails (e.g., duplicate anchors)
    """

def compose_all(stream, Loader=None):
    """
    Compose all YAML documents from stream into node trees.
    
    Parameters:
    - stream: YAML input containing multiple documents
    - Loader: Loader class to use for composition (defaults to SafeLoader)
    
    Yields:
    - Node objects representing each document structure
    
    Raises:
    - ComposerError: If node composition fails
    """

Serialization

Functions that convert node trees back to YAML events and text.

def serialize(
    node,
    stream=None,
    Dumper=None,
    *,
    canonical=None,
    indent=None,
    width=None,
    allow_unicode=None,
    line_break=None,
    encoding=None,
    explicit_start=None,
    explicit_end=None,
    version=None,
    tags=None
):
    """
    Serialize a node tree to YAML events.
    
    Parameters:
    - node: Node object to serialize
    - stream: Output stream (if None, returns string)
    - Dumper: Dumper class to use (defaults to SafeDumper)
    - (formatting parameters same as dump function)
    
    Returns:
    - YAML string if stream is None, otherwise None
    
    Raises:
    - SerializerError: If serialization fails
    """

def serialize_all(
    nodes,
    stream=None,
    Dumper=None,
    *,
    canonical=None,
    indent=None,
    width=None,
    allow_unicode=None,
    line_break=None,
    encoding=None,
    explicit_start=None,
    explicit_end=None,
    version=None,
    tags=None
):
    """
    Serialize multiple node trees to YAML events.
    
    Parameters:
    - nodes: Sequence of Node objects to serialize
    - stream: Output stream (if None, returns string)
    - Dumper: Dumper class to use (defaults to SafeDumper)
    - (formatting parameters same as dump function)
    
    Returns:
    - YAML string with multiple documents if stream is None, otherwise None
    
    Raises:
    - SerializerError: If serialization fails
    """

Event Emission

Function for converting YAML events directly to text output.

def emit(
    events,
    stream=None,
    Dumper=None,
    canonical=None,
    indent=None,
    width=None,
    allow_unicode=None,
    line_break=None,
    encoding=None,
    explicit_start=None,
    explicit_end=None,
    version=None,
    tags=None
):
    """
    Emit YAML events as text.
    
    Parameters:
    - events: Iterable of Event objects
    - stream: Output stream (if None, returns string)
    - Dumper: Dumper class to use (defaults to SafeDumper)
    - (formatting parameters same as dump function)
    
    Returns:
    - YAML string if stream is None, otherwise None
    
    Raises:
    - EmitterError: If emission fails
    """

Usage Examples

Token Scanning

import yaml

yaml_input = """
name: John Doe
age: 30
skills:
  - Python
  - YAML
"""

print("Tokens:")
for token in yaml.scan(yaml_input):
    print(f"  {type(token).__name__}: {token}")

# Output shows tokens like:
# StreamStartToken
# BlockMappingStartToken
# KeyToken
# ScalarToken: name
# ValueToken
# ScalarToken: John Doe
# KeyToken
# ScalarToken: age
# ...

Event Parsing

import yaml

yaml_input = """
users:
  - name: Alice
    role: admin
  - name: Bob
    role: user
"""

print("Events:")
for event in yaml.parse(yaml_input):
    print(f"  {type(event).__name__}: {event}")

# Output shows events like:
# StreamStartEvent
# DocumentStartEvent
# MappingStartEvent
# ScalarEvent: users
# SequenceStartEvent
# MappingStartEvent
# ScalarEvent: name
# ScalarEvent: Alice
# ...

Node Composition

import yaml

yaml_input = """
config: &default_config
  debug: false
  timeout: 30

development:
  <<: *default_config
  debug: true

production:
  <<: *default_config
  timeout: 60
"""

# Compose into node tree
root_node = yaml.compose(yaml_input)
print(f"Root node: {type(root_node).__name__}")
print(f"Tag: {root_node.tag}")
print(f"Value type: {type(root_node.value)}")

# Inspect node structure
if hasattr(root_node, 'value'):
    for key_node, value_node in root_node.value:
        print(f"Key: {key_node.value} -> Value type: {type(value_node).__name__}")

Custom Processing Pipeline

import yaml

def custom_yaml_processor(yaml_input):
    """Custom YAML processing with intermediate inspection."""
    
    print("=== SCANNING ===")
    tokens = list(yaml.scan(yaml_input))
    print(f"Found {len(tokens)} tokens")
    
    print("\n=== PARSING ===")
    events = list(yaml.parse(yaml_input))
    print(f"Found {len(events)} events")
    
    print("\n=== COMPOSING ===")
    node = yaml.compose(yaml_input)
    print(f"Root node: {type(node).__name__} with tag {node.tag}")
    
    print("\n=== CONSTRUCTING ===")
    # Use the composed node to construct Python objects
    loader = yaml.SafeLoader(yaml_input)
    try:
        loader.get_single_data()  # This actually constructs the objects
        constructed_data = loader.construct_document(node)
        print(f"Constructed: {constructed_data}")
    finally:
        loader.dispose()
    
    return constructed_data

# Test custom processor
yaml_input = """
name: Test Document
items:
  - id: 1
    value: alpha
  - id: 2
    value: beta
"""

result = custom_yaml_processor(yaml_input)

Manual Event Generation

import yaml

def create_yaml_events():
    """Create YAML events manually and emit them."""
    
    events = [
        yaml.StreamStartEvent(),
        yaml.DocumentStartEvent(),
        yaml.MappingStartEvent(anchor=None, tag=None, implicit=True),
        
        # Key: name
        yaml.ScalarEvent(anchor=None, tag=None, implicit=(True, True), value='name'),
        yaml.ScalarEvent(anchor=None, tag=None, implicit=(True, True), value='Manual YAML'),
        
        # Key: items  
        yaml.ScalarEvent(anchor=None, tag=None, implicit=(True, True), value='items'),
        yaml.SequenceStartEvent(anchor=None, tag=None, implicit=True),
        
        # First item
        yaml.ScalarEvent(anchor=None, tag=None, implicit=(True, True), value='item1'),
        yaml.ScalarEvent(anchor=None, tag=None, implicit=(True, True), value='item2'),
        
        yaml.SequenceEndEvent(),
        yaml.MappingEndEvent(),
        yaml.DocumentEndEvent(),
        yaml.StreamEndEvent()
    ]
    
    # Emit events to YAML text
    yaml_output = yaml.emit(events)
    return yaml_output

generated_yaml = create_yaml_events()
print("Generated YAML:")
print(generated_yaml)

# Verify by parsing back
data = yaml.safe_load(generated_yaml)
print(f"Parsed back: {data}")

Node Tree Manipulation

import yaml

def modify_yaml_tree(yaml_input):
    """Modify YAML at the node level before construction."""
    
    # Compose to node tree
    root_node = yaml.compose(yaml_input)
    
    # Find and modify specific nodes
    if isinstance(root_node, yaml.MappingNode):
        for key_node, value_node in root_node.value:
            if (isinstance(key_node, yaml.ScalarNode) and 
                key_node.value == 'debug'):
                # Change debug value to True
                if isinstance(value_node, yaml.ScalarNode):
                    value_node.value = 'true'
                    value_node.tag = 'tag:yaml.org,2002:bool'
            
            elif (isinstance(key_node, yaml.ScalarNode) and 
                  key_node.value == 'version'):
                # Increment version number
                if isinstance(value_node, yaml.ScalarNode):
                    try:
                        current_version = float(value_node.value)
                        value_node.value = str(current_version + 0.1)
                    except ValueError:
                        pass
    
    # Serialize modified tree back to YAML
    modified_yaml = yaml.serialize(root_node)
    return modified_yaml

# Test node manipulation
original_yaml = """
app_name: MyApp
version: 1.0
debug: false
features:
  - auth
  - logging
"""

print("Original YAML:")
print(original_yaml)

modified_yaml = modify_yaml_tree(original_yaml)
print("\nModified YAML:")
print(modified_yaml)

# Parse modified YAML to verify
modified_data = yaml.safe_load(modified_yaml)
print(f"\nParsed modified data: {modified_data}")

Stream Processing

import yaml
from io import StringIO

def stream_yaml_processing():
    """Process YAML using streams for memory efficiency."""
    
    # Create large YAML document
    yaml_content = """
    users:
    """
    
    # Add many users
    for i in range(1000):
        yaml_content += f"""
      - id: {i}
        name: User{i}
        email: user{i}@example.com
    """
    
    print(f"Processing YAML with {len(yaml_content)} characters")
    
    # Process with streaming
    stream = StringIO(yaml_content)
    
    # Use compose to build node tree without full construction
    root_node = yaml.compose(stream)
    print(f"Composed node tree: {type(root_node).__name__}")
    
    # Process the tree structure without loading all data into memory
    user_count = 0
    if isinstance(root_node, yaml.MappingNode):
        for key_node, value_node in root_node.value:
            if (isinstance(key_node, yaml.ScalarNode) and 
                key_node.value == 'users' and
                isinstance(value_node, yaml.SequenceNode)):
                user_count = len(value_node.value)
                break
    
    print(f"Found {user_count} users without loading all data")
    
    # Now selectively construct only what we need
    stream.seek(0)  # Reset stream
    full_data = yaml.safe_load(stream)
    first_user = full_data['users'][0] if full_data['users'] else None
    last_user = full_data['users'][-1] if full_data['users'] else None
    
    print(f"First user: {first_user}")
    print(f"Last user: {last_user}")

stream_yaml_processing()

Install with Tessl CLI