Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
packages: List[PackageSpec]
@classmethod
def from_project(cls, project):
return cls(name=project.project_name,
packages=project.packages.packages)
@dataclass
class Downloads(ExtensibleJsonSchemaMixin, Replaceable):
tarball: str
@dataclass
class RegistryPackageMetadata(
ExtensibleJsonSchemaMixin,
ProjectPackageMetadata,
):
downloads: Downloads
# A list of all the reserved words that packages may not have as names.
BANNED_PROJECT_NAMES = {
'_sql_results',
'adapter',
'api',
'column',
'config',
'context',
'database',
'env',
'env_var',
packages: List[PackageSpec]
@dataclass
class ProjectPackageMetadata:
name: str
packages: List[PackageSpec]
@classmethod
def from_project(cls, project):
return cls(name=project.project_name,
packages=project.packages.packages)
@dataclass
class Downloads(ExtensibleJsonSchemaMixin, Replaceable):
tarball: str
@dataclass
class RegistryPackageMetadata(
ExtensibleJsonSchemaMixin,
ProjectPackageMetadata,
):
downloads: Downloads
# A list of all the reserved words that packages may not have as names.
BANNED_PROJECT_NAMES = {
'_sql_results',
'adapter',
'api',
@property
def handle(self):
return self._handle
@handle.setter
def handle(self, value):
self._handle = value
# see https://github.com/python/mypy/issues/4717#issuecomment-373932080
# and https://github.com/python/mypy/issues/5374
# for why we have type: ignore. Maybe someday dataclasses + abstract classes
# will work.
@dataclass
class Credentials( # type: ignore
ExtensibleJsonSchemaMixin,
Replaceable,
metaclass=abc.ABCMeta
):
database: str
schema: str
_ALIASES: ClassVar[Dict[str, str]] = field(default={}, init=False)
@abc.abstractproperty
def type(self) -> str:
raise NotImplementedError(
'type not implemented for base credentials class'
)
def connection_info(self) -> Iterable[Tuple[str, Any]]:
"""Return an ordered iterator of key/value pairs for pretty-printing.
"""
from dbt.utils import translate_aliases
Identifier = NewType('Identifier', str)
register_pattern(Identifier, r'^[A-Za-z_][A-Za-z0-9_]+$')
class ConnectionState(StrEnum):
INIT = 'init'
OPEN = 'open'
CLOSED = 'closed'
FAIL = 'fail'
@dataclass(init=False)
class Connection(ExtensibleJsonSchemaMixin, Replaceable):
type: Identifier
name: Optional[str]
_credentials: JsonSchemaMixin = None # underscore to prevent serialization
state: ConnectionState = ConnectionState.INIT
transaction_open: bool = False
_handle: Optional[Any] = None # underscore to prevent serialization
def __init__(
self,
type: Identifier,
name: Optional[str],
credentials: JsonSchemaMixin,
state: ConnectionState = ConnectionState.INIT,
transaction_open: bool = False,
handle: Optional[Any] = None,
) -> None:
filter: Optional[str] = None
def status(self, age: float) -> FreshnessStatus:
if self.error_after and self.error_after.exceeded(age):
return FreshnessStatus.Error
elif self.warn_after and self.warn_after.exceeded(age):
return FreshnessStatus.Warn
else:
return FreshnessStatus.Pass
def __bool__(self):
return self.warn_after is not None or self.error_after is not None
@dataclass
class AdditionalPropertiesAllowed(ExtensibleJsonSchemaMixin):
_extra: Dict[str, Any] = field(default_factory=dict)
@property
def extra(self):
return self._extra
@classmethod
def from_dict(cls, data, validate=True):
self = super().from_dict(data=data, validate=validate)
keys = self.to_dict(validate=False, omit_none=False)
for key, value in data.items():
if key not in keys:
self._extra[key] = value
return self
def to_dict(self, omit_none=True, validate=False):
packages: List[PackageSpec] = field(default_factory=list)
query_comment: Optional[Union[str, NoValue]] = NoValue()
@classmethod
def from_dict(cls, data, validate=True):
result = super().from_dict(data, validate=validate)
if result.name in BANNED_PROJECT_NAMES:
raise ValidationError(
'Invalid project name: {} is a reserved word'
.format(result.name)
)
return result
@dataclass
class UserConfig(ExtensibleJsonSchemaMixin, Replaceable):
send_anonymous_usage_stats: bool = DEFAULT_SEND_ANONYMOUS_USAGE_STATS
use_colors: bool = DEFAULT_USE_COLORS
partial_parse: Optional[bool] = None
printer_width: Optional[int] = None
def set_values(self, cookie_dir):
if self.send_anonymous_usage_stats:
tracking.initialize_tracking(cookie_dir)
else:
tracking.do_not_track()
if self.use_colors:
printer.use_colors()
if self.printer_width:
printer.printer_width(self.printer_width)