Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_invalid(self):
with self.assertRaises(ValidationError):
GitPackageContract.from_dict(
{'git': 'http://example.com', 'version': '0.0.1'}
)
def assert_fails_validation(self, dct, cls=None):
if cls is None:
cls = self.ContractType
with self.assertRaises(ValidationError):
cls.from_dict(dct)
}
format_stats will convert the dict into a StatsDict with keys of 'encoded'
and 'size'.
"""
stats_collector: StatsDict = {}
base_keys = {k.split(':')[0] for k in stats}
for key in base_keys:
dct: PrimitiveDict = {'id': key}
for subkey in ('label', 'value', 'description', 'include'):
dct[subkey] = stats['{}:{}'.format(key, subkey)]
try:
stats_item = StatsItem.from_dict(dct)
except ValidationError:
continue
if stats_item.include:
stats_collector[key] = stats_item
# we always have a 'has_stats' field, it's never included
has_stats = StatsItem(
id='has_stats',
label='Has Stats?',
value=len(stats_collector) > 0,
description='Indicates whether there are statistics for this table',
include=False,
)
stats_collector['has_stats'] = has_stats
return stats_collector
@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})\.