TagSet Class

Overview

The TagSet class provides a collection of key-value pairs where keys are labels and values are SerializedData objects. TagSets are used extensively throughout the Keychain system for metadata storage, credential claims, consensus participant definitions, and variable tracking.

Package: keychain.core.tag_set

from keychain.core.tag_set import TagSet

Class Definition

class TagSet(KeychainObject):
    """A tag set is a collection of key-value pairs where keys are labels
    and values are SerializedData objects."""

Constructor

def __init__(self, c_pointer: Optional[ctypes.c_void_p] = None) -> None

Initialize a TagSet object. If no C pointer is provided, creates an empty tag set.

Parameters:

  • c_pointer (ctypes.c_void_p, optional) - Existing C pointer to wrap

Example:

# Create empty tag set
tags = TagSet()

# Wrap existing C pointer (typically from library functions)
tags = TagSet(c_pointer=existing_pointer)

Key-Value Operations

get_tag_value(namespace, name)

def get_tag_value(self, namespace: str, name: str) -> SerializedData

Get tag value by namespace and name.

Parameters:

  • namespace (str) - Tag namespace

  • name (str) - Tag name

Returns: SerializedData value for the tag

Raises: KeychainValidationError if tag is not found

Example:

try:
    value = tags.get_tag_value("employee", "department")
    print(f"Department: {value.content}")
except KeychainValidationError:
    print("Tag not found")

set_tag_value(namespace, name, value)

def set_tag_value(self, namespace: str, name: str, value: SerializedData) -> None

Set tag value for the specified namespace and name.

Parameters:

  • namespace (str) - Tag namespace

  • name (str) - Tag name

  • value (SerializedData) - SerializedData value to set

Example:

from keychain.core.serialized_data import SerializedData
from keychain.constants import DataType

# Create serialized data value
dept_value = SerializedData.from_string("Engineering", DataType.UTF8_STRING)

# Set tag value
tags.set_tag_value("employee", "department", dept_value)

Dictionary-Style Access

getitem(key)

def __getitem__(self, key: Tuple[str, str]) -> SerializedData

Get item using (namespace, name) tuple key.

Parameters:

  • key (Tuple[str, str]) - Tuple of (namespace, name)

Returns: SerializedData value

Raises: KeychainValidationError if key format is invalid or tag not found

setitem(key, value)

def __setitem__(self, key: Tuple[str, str], value: SerializedData) -> None

Set item using (namespace, name) tuple key.

Parameters:

  • key (Tuple[str, str]) - Tuple of (namespace, name)

  • value (SerializedData) - Value to set

Raises: KeychainValidationError if key format is invalid

Example:

from keychain.core.serialized_data import SerializedData
from keychain.constants import DataType

# Dictionary-style access
tags[("employee", "name")] = SerializedData.from_string("Alice Smith", DataType.UTF8_STRING)
tags[("employee", "id")] = SerializedData.from_string("12345", DataType.UTF8_STRING)

# Retrieve values
name_data = tags[("employee", "name")]
print(f"Employee: {name_data.content}")

Comparison Methods

eq(other)

def __eq__(self, other: object) -> bool

Check equality with another TagSet.

Parameters:

  • other (TagSet) - TagSet to compare with

Returns: True if tag sets are equal

Example:

tags1 = TagSet()
tags2 = TagSet()

# Set same values
value = SerializedData.from_string("test", DataType.UTF8_STRING)
tags1.set_tag_value("", "key", value)
tags2.set_tag_value("", "key", value)

if tags1 == tags2:
    print("Tag sets are equal")

Example: Employee Information TagSet

from keychain.core.tag_set import TagSet
from keychain.core.serialized_data import SerializedData
from keychain.constants import DataType

def create_employee_tags(employee_id: str, name: str, department: str, role: str) -> TagSet:
    """Create a TagSet with employee information."""

    tags = TagSet()

    # Add employee information using empty namespace
    tags.set_tag_value("", "employee_id",
                      SerializedData.from_string(employee_id, DataType.UTF8_STRING))
    tags.set_tag_value("", "name",
                      SerializedData.from_string(name, DataType.UTF8_STRING))
    tags.set_tag_value("", "department",
                      SerializedData.from_string(department, DataType.UTF8_STRING))
    tags.set_tag_value("", "role",
                      SerializedData.from_string(role, DataType.UTF8_STRING))

    # Add additional metadata with namespace
    tags.set_tag_value("metadata", "created_by",
                      SerializedData.from_string("hr_system", DataType.UTF8_STRING))

    import time
    timestamp = str(int(time.time() * 1000))
    tags.set_tag_value("metadata", "created_at",
                      SerializedData.from_string(timestamp, DataType.UTF8_STRING))

    return tags

# Usage example
employee_tags = create_employee_tags("12345", "Alice Johnson", "Engineering", "Senior Developer")

# Access values using dictionary-style notation
name_data = employee_tags[("", "name")]
dept_data = employee_tags[("", "department")]

print(f"Employee: {name_data.content}")
print(f"Department: {dept_data.content}")

# Access values using method calls
role_data = employee_tags.get_tag_value("", "role")
created_by = employee_tags.get_tag_value("metadata", "created_by")

print(f"Role: {role_data.content}")
print(f"Created by: {created_by.content}")

Example: Credential Claims TagSet

from keychain.core.tag_set import TagSet
from keychain.core.serialized_data import SerializedData
from keychain.constants import DataType
import json

def create_credential_claims(claims_dict: dict) -> TagSet:
    """Create a TagSet for credential claims from a dictionary."""

    claims = TagSet()

    for key, value in claims_dict.items():
        # Convert various Python types to serialized data
        if isinstance(value, str):
            serialized_value = SerializedData.from_string(value, DataType.UTF8_STRING)
        elif isinstance(value, int):
            serialized_value = SerializedData.from_string(str(value), DataType.UTF8_STRING)
        elif isinstance(value, bool):
            serialized_value = SerializedData.from_string(str(value).lower(), DataType.UTF8_STRING)
        elif isinstance(value, (dict, list)):
            # Serialize complex objects as JSON
            json_str = json.dumps(value)
            serialized_value = SerializedData.from_string(json_str, DataType.JSON)
        else:
            # Convert other types to string
            serialized_value = SerializedData.from_string(str(value), DataType.UTF8_STRING)

        claims.set_tag_value("", key, serialized_value)

    return claims

# Usage example
claims_data = {
    "name": "Bob Smith",
    "employee_id": "67890",
    "department": "Marketing",
    "role": "Manager",
    "clearance_level": "Standard",
    "permissions": ["read", "write", "approve"],
    "active": True,
    "start_date": "2023-01-15"
}

claims = create_credential_claims(claims_data)

# Verify claims were set correctly
for key in claims_data.keys():
    value_data = claims[("", key)]
    print(f"{key}: {value_data.content}")

Example: Consensus Quorum TagSet

from keychain.core.tag_set import TagSet
from keychain.core.serialized_data import SerializedData
from keychain.constants import DataType

def create_consensus_quorum(participants: dict) -> TagSet:
    """Create a TagSet defining consensus participants and their roles."""

    quorum = TagSet()

    # Set overall quorum parameters
    quorum.set_tag_value("", "algorithm",
                        SerializedData.from_string("majority", DataType.UTF8_STRING))
    quorum.set_tag_value("", "threshold",
                        SerializedData.from_string("2", DataType.UTF8_STRING))

    # Add participants with their roles
    for participant_id, role in participants.items():
        quorum.set_tag_value("participants", participant_id,
                           SerializedData.from_string(role, DataType.UTF8_STRING))

    return quorum

# Usage example
participants = {
    "did:keychain:alice.company.com": "approver",
    "did:keychain:bob.company.com": "reviewer",
    "did:keychain:charlie.company.com": "auditor"
}

quorum = create_consensus_quorum(participants)

# Check quorum parameters
algorithm = quorum[("", "algorithm")]
threshold = quorum[("", "threshold")]

print(f"Consensus algorithm: {algorithm.content}")
print(f"Threshold: {threshold.content}")

# Check participant roles
for participant_id in participants.keys():
    try:
        role_data = quorum.get_tag_value("participants", participant_id)
        print(f"{participant_id}: {role_data.content}")
    except KeychainValidationError:
        print(f"Role not found for {participant_id}")

Example: Variable Tracking TagSet

from keychain.core.tag_set import TagSet
from keychain.core.serialized_data import SerializedData
from keychain.constants import DataType

class WorkflowVariables:
    """Helper class for managing workflow variables in TagSets."""

    def __init__(self):
        self.variables = TagSet()

    def set_string_variable(self, name: str, value: str) -> None:
        """Set a string variable."""
        data = SerializedData.from_string(value, DataType.UTF8_STRING)
        self.variables.set_tag_value("vars", name, data)

    def set_number_variable(self, name: str, value: int) -> None:
        """Set a numeric variable."""
        data = SerializedData.from_string(str(value), DataType.UTF8_STRING)
        self.variables.set_tag_value("vars", name, data)

    def set_status_variable(self, name: str, status: str) -> None:
        """Set a status variable."""
        data = SerializedData.from_string(status, DataType.UTF8_STRING)
        self.variables.set_tag_value("status", name, data)

    def get_string_variable(self, name: str) -> str:
        """Get a string variable."""
        data = self.variables.get_tag_value("vars", name)
        return data.content

    def get_number_variable(self, name: str) -> int:
        """Get a numeric variable."""
        data = self.variables.get_tag_value("vars", name)
        return int(data.content)

    def get_status_variable(self, name: str) -> str:
        """Get a status variable."""
        data = self.variables.get_tag_value("status", name)
        return data.content

    def get_tagset(self) -> TagSet:
        """Get the underlying TagSet."""
        return self.variables

# Usage example
workflow_vars = WorkflowVariables()

# Set workflow variables
workflow_vars.set_string_variable("project_name", "New Product Launch")
workflow_vars.set_string_variable("lead_engineer", "alice.company.com")
workflow_vars.set_number_variable("iteration", 3)
workflow_vars.set_number_variable("budget", 50000)

# Set status variables
workflow_vars.set_status_variable("approval_status", "pending")
workflow_vars.set_status_variable("review_status", "in_progress")

# Retrieve variables
project = workflow_vars.get_string_variable("project_name")
iteration = workflow_vars.get_number_variable("iteration")
approval_status = workflow_vars.get_status_variable("approval_status")

print(f"Project: {project}")
print(f"Iteration: {iteration}")
print(f"Approval Status: {approval_status}")

# Get the TagSet for use with other Keychain operations
tags = workflow_vars.get_tagset()

Error Handling

TagSet operations can raise KeychainValidationError exceptions:

Example:

from keychain.exceptions import KeychainValidationError
from keychain.core.tag_set import TagSet

tags = TagSet()

try:
    # Invalid key format
    value = tags[("single_element",)]  # Tuple must have exactly 2 elements
except KeychainValidationError as e:
    print(f"Key format error: {e}")

try:
    # Tag not found
    value = tags.get_tag_value("nonexistent", "tag")
except KeychainValidationError as e:
    print(f"Tag not found: {e}")

try:
    # Invalid key type
    value = tags["string_key"]  # Must be tuple
except KeychainValidationError as e:
    print(f"Invalid key type: {e}")

Best Practices

  1. Use meaningful namespaces: Organize tags with descriptive namespaces to avoid conflicts

  2. Consistent data types: Use appropriate DataType values when creating SerializedData

  3. Error handling: Always wrap tag operations in try-catch blocks

  4. Empty namespace: Use empty string "" for the default namespace

  5. Key naming: Use descriptive, consistent key names

Example:

def safe_tag_operations():
    """Demonstrate safe TagSet usage patterns."""

    tags = TagSet()

    try:
        # Use meaningful namespaces
        user_data = SerializedData.from_string("alice", DataType.UTF8_STRING)
        tags.set_tag_value("user", "name", user_data)

        system_data = SerializedData.from_string("v1.2.3", DataType.UTF8_STRING)
        tags.set_tag_value("system", "version", system_data)

        # Safe access with error handling
        try:
            name = tags.get_tag_value("user", "name")
            print(f"User name: {name.content}")
        except KeychainValidationError:
            print("User name not found")

        # Dictionary-style access with proper tuple keys
        try:
            version = tags[("system", "version")]
            print(f"System version: {version.content}")
        except KeychainValidationError:
            print("System version not found")

        return tags

    except Exception as e:
        print(f"TagSet operation failed: {e}")
        return None

See Also