Skip to content

_register

_register

Dynamic property registration for conformed properties.

This module provides utilities to automatically register conformed properties as @property methods on the DomoStream class.

register_conformed_properties

register_conformed_properties(
    cls, properties_registry: dict
)

Dynamically register conformed properties onto a class.

Parameters:

Name Type Description Default
cls

Class to register properties on (DomoStream)

required
properties_registry dict

Dict of ConformedProperty objects (CONFORMED_PROPERTIES)

required
Example

from .stream_configs._conformed import CONFORMED_PROPERTIES register_conformed_properties(DomoStream, CONFORMED_PROPERTIES)

Source code in src/crew_dcs/classes/DomoDataset/stream_configs/_register.py
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def register_conformed_properties(cls, properties_registry: dict):
    """Dynamically register conformed properties onto a class.

    Args:
        cls: Class to register properties on (DomoStream)
        properties_registry: Dict of ConformedProperty objects (CONFORMED_PROPERTIES)

    Example:
        >>> from .stream_configs._conformed import CONFORMED_PROPERTIES
        >>> register_conformed_properties(DomoStream, CONFORMED_PROPERTIES)
    """
    # Map of property names to their display names
    property_name_map = {
        "query": "sql",
        "database": "database",
        "schema": "schema",
        "warehouse": "warehouse",
        "table": "table",
        "report_id": "report_id",
        "spreadsheet": "spreadsheet",
        "bucket": "bucket",
        "dataset_id": "dataset_id",
        "file_url": "file_url",
        "host": "host",
        "port": "port",
        "custom_query": None,  # Internal, not exposed as property
    }

    for prop_name, conf_prop in properties_registry.items():
        # Get the public property name
        public_name = property_name_map.get(prop_name)

        # Skip if no public name (internal properties)
        if public_name is None:
            continue

        # Skip if property already exists (manual override)
        if hasattr(cls, public_name):
            continue

        # Create docstring
        doc = f"""Get {conf_prop.description or prop_name} from stream configuration.

        {conf_prop.description or "Configuration parameter"}

        Supported providers: {", ".join(conf_prop.supported_providers[:3])}
        {"..." if len(conf_prop.supported_providers) > 3 else ""}
        ({len(conf_prop.supported_providers)} total)

        Returns:
            Value from configuration or None if not available

        Example:
            >>> stream = await DomoStream.get_by_id(auth, stream_id)
            >>> print(stream.{public_name})
        """

        # Create and register the property
        prop = _create_property_getter(prop_name, doc)
        setattr(cls, public_name, prop)

register_conformed_property

register_conformed_property(
    cls,
    property_name: str,
    public_name: str | None = None,
    docstring: str | None = None,
)

Register a single conformed property onto a class.

Decorator-based approach for adding individual properties.

Parameters:

Name Type Description Default
cls

Class to register property on

required
property_name str

Name in CONFORMED_PROPERTIES registry

required
public_name str | None

Name of the property on the class (defaults to property_name)

None
docstring str | None

Optional custom docstring

None
Example

@register_conformed_property(DomoStream, "query", "sql") ... class DomoStream: ... pass

Source code in src/crew_dcs/classes/DomoDataset/stream_configs/_register.py
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def register_conformed_property(
    cls,
    property_name: str,
    public_name: str | None = None,
    docstring: str | None = None,
):
    """Register a single conformed property onto a class.

    Decorator-based approach for adding individual properties.

    Args:
        cls: Class to register property on
        property_name: Name in CONFORMED_PROPERTIES registry
        public_name: Name of the property on the class (defaults to property_name)
        docstring: Optional custom docstring

    Example:
        >>> @register_conformed_property(DomoStream, "query", "sql")
        ... class DomoStream:
        ...     pass
    """
    public_name = public_name or property_name

    # Skip if already exists
    if hasattr(cls, public_name):
        return cls

    doc = docstring or f"Get {property_name} from stream configuration."

    prop = _create_property_getter(property_name, doc)
    setattr(cls, public_name, prop)

    return cls