Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Finalize validations for custom granularities #370

Merged
merged 8 commits into from
Dec 2, 2024
17 changes: 11 additions & 6 deletions dbt_semantic_interfaces/validations/validator_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@
from datetime import date
from enum import Enum
from typing import (
Any,
Callable,
Dict,
Generic,
Expand All @@ -20,6 +19,7 @@
)

import click
from typing_extensions import ParamSpec

from dbt_semantic_interfaces.implementations.base import FrozenBaseModel
from dbt_semantic_interfaces.protocols import Metadata, SemanticManifestT, SemanticModel
Expand All @@ -35,6 +35,8 @@
ValidationContextJSON = Dict[str, Union[str, int, None]]
ValidationIssueJSON = Dict[str, Union[str, int, ValidationContextJSON]]

P = ParamSpec("P")


class ValidationIssueLevel(Enum):
"""Categorize the issues found while validating a semantic manifest."""
Expand Down Expand Up @@ -384,18 +386,21 @@ def generate_exception_issue(
)


def _func_args_to_string(*args: Any, **kwargs: Any) -> str: # type: ignore
def _func_args_to_string(*args: P.args, **kwargs: P.kwargs) -> str: # type: ignore
return f"positional args: {args}, key word args: {kwargs}"


def validate_safely(whats_being_done: str) -> Callable:
def validate_safely(
whats_being_done: str,
) -> Callable[[Callable[P, Sequence[ValidationIssue]]], Callable[P, Sequence[ValidationIssue]]]:
"""Decorator to safely run validation checks."""

def decorator_check_element_safely(func: Callable) -> Callable: # noqa
def decorator_check_element_safely(
func: Callable[P, Sequence[ValidationIssue]]
) -> Callable[P, Sequence[ValidationIssue]]:
@functools.wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> List[ValidationIssue]: # type: ignore
def wrapper(*args: P.args, **kwargs: P.kwargs) -> Sequence[ValidationIssue]: # type: ignore
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Noooice!

"""Safely run a check on model elements."""
issues: List[ValidationIssue]
try:
issues = func(*args, **kwargs)
except Exception as e:
Expand Down