Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@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]
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,
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'))
@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
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
'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))
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})\.
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):
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
@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
)