gqueries

gqueries

Wraps a dict of queries and answers

Classes

Gqueries

Gqueries(**data)

Bases: Base

We cannot validate yet - as we'd need a service connected to the main gquery endpoint

Source code in src/pyetm/models/base.py

def __init__(self, **data: Any) -> None:
    """
    Initialize the model, converting validation errors to warnings.
    """
    super(BaseModel, self).__setattr__("__pydantic_private__", {})

    # Initialize all private attributes with their defaults
    private_dict: Dict[str, Any] = self.__pydantic_private__  # type: ignore[assignment]
    for attr_name, attr_info in self.__class__.__private_attributes__.items():
        if (
            hasattr(attr_info, "default_factory")
            and attr_info.default_factory is not None
        ):
            # Call factory - signature varies between pydantic versions
            private_dict[attr_name] = attr_info.default_factory()
        elif hasattr(attr_info, "default"):
            private_dict[attr_name] = attr_info.default
        else:
            private_dict[attr_name] = None

    try:
        super().__init__(**data)
    except ValidationError as e:
        # Check if data is None or empty - this indicates API error
        if not data or data is None:
            # Re-raise with clearer message
            raise ValueError(
                f"Cannot create {self.__class__.__name__} with empty data. "
                "This usually indicates an authentication or API error."
            ) from e

        # If validation fails, create model without validation and collect warnings
        # Use model_construct to bypass validation
        temp_instance = self.__class__.model_construct(**data)

        # Copy the constructed data to this instance
        for field_name, field_value in temp_instance.__dict__.items():
            if not field_name.startswith("_"):
                object.__setattr__(self, field_name, field_value)

        # Ensure required Pydantic slot attributes exist to prevent AttributeError
        for slot in ("__pydantic_fields_set__", "__pydantic_extra__"):
            try:
                value = object.__getattribute__(temp_instance, slot)
                object.__setattr__(self, slot, value)
            except AttributeError:
                # Initialize missing slot attributes with defaults
                if slot == "__pydantic_extra__":
                    object.__setattr__(self, slot, {})
                elif slot == "__pydantic_fields_set__":
                    object.__setattr__(self, slot, set())

        # Convert validation errors to warnings
        for error in e.errors():
            field_path = ".".join(str(part) for part in error.get("loc", []))
            message = error.get("msg", "Validation failed")
            self._warning_collector.add(field_path, message, "error")

Functions

update

update(json)

Updates the values with a JSON response from the API.

Source code in src/pyetm/models/gqueries.py

def update(self, json: dict[str, Any]) -> None:
    """
    Updates the values with a JSON response from the API.
    """
    processed_json: dict[str, Any] = {}

    for key, value in json.items():
        processed_json[key] = value

    self.query_dict.update(processed_json)

get

get(key)

Returns the query value if set, otherwise returns None

Source code in src/pyetm/models/gqueries.py

def get(self, key: str) -> Any:
    """
    Returns the query value if set, otherwise returns None
    """
    return self.query_dict.get(key, None)

add

add(*query_keys)

Add more queries to be requested

Source code in src/pyetm/models/gqueries.py

def add(self, *query_keys: str) -> None:
    """
    Add more queries to be requested
    """
    self.query_dict.update({q: None for q in query_keys if q not in self.query_dict.keys()})

remove

remove(*query_keys)

Remove specific query keys from the collection.

Invalid query keys are rejected with warnings. Warnings are automatically displayed for non-existent keys. Warnings from previous removals are cleared to show only current operation issues.

Parameters:

Name	Type	Description	Default
query_keys	str	Query keys to remove from the collection	()

Source code in src/pyetm/models/gqueries.py

def remove(self, *query_keys: str) -> None:
    """
    Remove specific query keys from the collection.

    Invalid query keys are rejected with warnings.
    Warnings are automatically displayed for non-existent keys.
    Warnings from previous removals are cleared to show only current operation issues.

    Args:
        query_keys: Query keys to remove from the collection
    """
    # Auto-clear stale warnings from previous removals
    self.warnings.clear()

    for key in query_keys:
        if key in self.query_dict:
            del self.query_dict[key]
        else:
            self.add_warning(key, f"Query '{key}' not found in collection")

    # Auto-display warnings if any exist
    if len(self.warnings) > 0:
        self.auto_show_warnings()

clear

clear()

Remove all queries from the collection. Also clears any accumulated warnings.

Source code in src/pyetm/models/gqueries.py

def clear(self) -> None:
    """
    Remove all queries from the collection.
    Also clears any accumulated warnings.
    """
    self.query_dict.clear()
    self.warnings.clear()