How to use the hologram.JsonSchemaMixin function in hologram

To help you get started, we’ve selected a few hologram examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github fishtown-analytics / dbt / core / dbt / contracts / results.py View on Github external
@dataclass
class StatsItem(JsonSchemaMixin):
    id: str
    label: str
    value: Primitive
    description: str
    include: bool


StatsDict = Dict[str, StatsItem]


@dataclass
class ColumnMetadata(JsonSchemaMixin):
    type: str
    comment: Optional[str]
    index: int
    name: str


ColumnMap = Dict[str, ColumnMetadata]


@dataclass
class TableMetadata(JsonSchemaMixin):
    type: str
    database: str
    schema: str
    name: str
    comment: Optional[str]
github fishtown-analytics / dbt / core / dbt / contracts / rpc.py View on Github external
compiled_sql: str
    node: CompileResultNode
    timing: List[TimingInfo]

    @property
    def error(self):
        return None


@dataclass
class RemoteExecutionResult(ExecutionResult, RemoteResult):
    pass


@dataclass
class ResultTable(JsonSchemaMixin):
    column_names: List[str]
    rows: List[Any]


@dataclass
class RemoteRunOperationResult(RemoteResult):
    success: bool


@dataclass
class RemoteRunResult(RemoteCompileResult):
    table: ResultTable


RPCResult = Union[
    RemoteCompileResult,
github fishtown-analytics / dbt / core / dbt / contracts / graph / parsed.py View on Github external
from dbt.contracts.util import Replaceable, list_str
from dbt.logger import GLOBAL_LOGGER as logger  # noqa
from dbt.node_types import NodeType


class SnapshotStrategy(StrEnum):
    Timestamp = 'timestamp'
    Check = 'check'


class All(StrEnum):
    All = 'all'


@dataclass
class Hook(JsonSchemaMixin, Replaceable):
    sql: str
    transaction: bool = True
    index: Optional[int] = None


def insensitive_patterns(*patterns: str):
    lowercased = []
    for pattern in patterns:
        lowercased.append(
            ''.join('[{}{}]'.format(s.upper(), s.lower()) for s in pattern)
        )
    return '^({})$'.format('|'.join(lowercased))


Severity = NewType('Severity', str)
register_pattern(Severity, insensitive_patterns('warn', 'error'))
github fishtown-analytics / dbt / core / dbt / contracts / results.py View on Github external
@dataclass
class SourceFreshnessOutput(JsonSchemaMixin):
    max_loaded_at: datetime
    snapshotted_at: datetime
    max_loaded_at_time_ago_in_s: Real
    state: FreshnessStatus
    criteria: FreshnessCriteria


SourceFreshnessRunResult = Union[SourceFreshnessOutput,
                                 SourceFreshnessRuntimeError]


@dataclass
class FreshnessRunOutput(JsonSchemaMixin, Writable):
    meta: FreshnessMetadata
    sources: Dict[str, SourceFreshnessRunResult]


Primitive = Union[bool, str, float, None]

CatalogKey = NamedTuple(
    'CatalogKey',
    [('database', str), ('schema', str), ('name', str)]
)


@dataclass
class StatsItem(JsonSchemaMixin):
    id: str
    label: str
github fishtown-analytics / dbt / core / dbt / contracts / rpc.py View on Github external
CatalogResults,
    ExecutionResult,
)
from dbt.exceptions import InternalException
from dbt.logger import LogMessage
from dbt.utils import restrict_to


TaskTags = Optional[Dict[str, Any]]
TaskID = uuid.UUID

# Inputs


@dataclass
class RPCParameters(JsonSchemaMixin):
    timeout: Optional[Real]
    task_tags: TaskTags


@dataclass
class RPCExecParameters(RPCParameters):
    name: str
    sql: str
    macros: Optional[str]


@dataclass
class RPCCompileParameters(RPCParameters):
    threads: Optional[int] = None
    models: Union[None, str, List[str]] = None
    exclude: Union[None, str, List[str]] = None
github fishtown-analytics / dbt / core / dbt / contracts / graph / parsed.py View on Github external
'Seeds should always have a seed_file_path'
            )

    @property
    def empty(self):
        """ Seeds are never empty"""
        return False


@dataclass
class TestConfig(NodeConfig):
    severity: Severity = Severity('error')


@dataclass
class TestMetadata(JsonSchemaMixin):
    namespace: Optional[str]
    name: str
    kwargs: Dict[str, Any]


@dataclass
class ParsedTestNode(ParsedNode):
    resource_type: NodeType = field(metadata={'restrict': [NodeType.Test]})
    column_name: Optional[str] = None
    config: TestConfig = field(default_factory=TestConfig)
    test_metadata: Optional[TestMetadata] = None


@dataclass(init=False)
class _SnapshotConfig(NodeConfig):
    unique_key: str = field(init=False, metadata=dict(init_required=True))
github fishtown-analytics / dbt / core / dbt / semver.py View on Github external
from hologram import JsonSchemaMixin
from hologram.helpers import StrEnum
from typing import Optional


class Matchers(StrEnum):
    GREATER_THAN = '>'
    GREATER_THAN_OR_EQUAL = '>='
    LESS_THAN = '<'
    LESS_THAN_OR_EQUAL = '<='
    EXACT = '='


@dataclass
class VersionSpecification(JsonSchemaMixin):
    major: Optional[str]
    minor: Optional[str]
    patch: Optional[str]
    prerelease: Optional[str]
    build: Optional[str]
    matcher: Matchers = Matchers.EXACT


_MATCHERS = r"(?P\>=|\>|\<|\<=|=)?"
_NUM_NO_LEADING_ZEROS = r"(0|[1-9][0-9]*)"
_ALPHA = r"[0-9A-Za-z-]*"
_ALPHA_NO_LEADING_ZEROS = r"(0|[1-9A-Za-z-][0-9A-Za-z-]*)"

_BASE_VERSION_REGEX = r"""
(?P{num_no_leading_zeros})\.
(?P{num_no_leading_zeros})\.
github fishtown-analytics / dbt / core / dbt / contracts / results.py View on Github external
def __len__(self):
        return len(self.results)

    def __iter__(self):
        return iter(self.results)

    def __getitem__(self, idx):
        return self.results[idx]


def _copykeys(src, keys, **updates):
    return {k: getattr(src, k) for k in keys}


@dataclass
class FreshnessCriteria(JsonSchemaMixin):
    warn_after: Time
    error_after: Time


class FreshnessErrorEnum(StrEnum):
    runtime_error = 'runtime error'


@dataclass
class SourceFreshnessRuntimeError(JsonSchemaMixin):
    error: str
    state: FreshnessErrorEnum


@dataclass
class SourceFreshnessOutput(JsonSchemaMixin):
github fishtown-analytics / dbt / core / dbt / task / generate.py View on Github external
if k.startswith(prefix)
    }


Primitive = Union[bool, str, float, None]
PrimitiveDict = Dict[str, Primitive]


Key = NamedTuple(
    'Key',
    [('database', str), ('schema', str), ('name', str)]
)


@dataclass
class StatsItem(JsonSchemaMixin):
    id: str
    label: str
    value: Primitive
    description: str
    include: bool


StatsDict = Dict[str, StatsItem]


@dataclass
class ColumnMetadata(JsonSchemaMixin):
    type: str
    comment: Optional[str]
    index: int
    name: str
github fishtown-analytics / dbt / core / dbt / contracts / graph / unparsed.py View on Github external
@dataclass
class ExternalTable(AdditionalPropertiesAllowed, Mergeable):
    location: Optional[str] = None
    file_format: Optional[str] = None
    row_format: Optional[str] = None
    tbl_properties: Optional[str] = None
    partitions: Optional[List[ExternalPartition]] = None

    def __bool__(self):
        return self.location is not None


@dataclass
class Quoting(JsonSchemaMixin, Mergeable):
    database: Optional[bool] = None
    schema: Optional[bool] = None
    identifier: Optional[bool] = None


@dataclass
class UnparsedSourceTableDefinition(ColumnDescription, NodeDescription):
    loaded_at_field: Optional[str] = None
    identifier: Optional[str] = None
    quoting: Quoting = field(default_factory=Quoting)
    freshness: Optional[FreshnessThreshold] = field(
        default_factory=FreshnessThreshold
    )
    external: Optional[ExternalTable] = field(
        default_factory=ExternalTable
    )