`snowmobile.core.cfg`¶

Full configuration object model; represents a parsed snowmobile.toml file.

Submodules¶

Package Contents¶

Classes¶

`Base`	Base class for object model parsed from `snowmobile.toml`.
`Connection`	[connection]
`Credentials`	[connection.credentials.credentials_alias]
`Loading`	[loading]
`Put`	[loading.put]
`Copy`	[loading.copy]
`SQL`	[sql] (snowmobile-ext.toml)
`Location`	[external-sources]
`QA`	[script.qa]
`Attributes`	[script.markdown.attributes]
`Markup`	[script.markup]
`Marker`	[script.markdown.attributes.markers]
`Pattern`	[script.patterns]
`Script`	[script]
`Wildcard`	[script.patterns.wildcards]

class snowmobile.core.cfg.Base¶

Bases: pydantic.BaseModel, snowmobile.core.cfg.base.Config

Base class for object model parsed from snowmobile.toml.

property configured_args(self) → Dict¶: Placeholder for configuration arguments of derived classes.

kwarg(self, arg_nm: str, arg_val: Any, arg_typ: Any) → Any¶: Compares a provided keyword argument to a configured keyword argument.

from_relative(self, obj: Any)¶: Updates current object’s attributes with those from a different instance of the same class.

from_dict(self, args: Dict)¶: Accept a dictionary of arguments and updates the current object as if it were instantiated with those arguments.

as_serializable(self, by_alias: bool = False)¶: Returns a dictionary in serializable form.

json(self, by_alias: bool = False, **kwargs) → str ¶: API-facing json serialization method.

class snowmobile.core.cfg.Connection(**data)¶

Bases: snowmobile.core.cfg.base.Base

[connection]

This includes the default_alias which is the set of credentials that snowmobile will authenticate with if creds is not explicitly passed.

default_alias¶

The set of credentials that is used if creds is not explicitly passed to snowmobile.connect on instantiation.

Type: str

creds¶

The name given to the set of credentials within the credentials block of the snowmobile.toml file (e.g. [credentials.creds] assigns an creds to a given set of credentials.

Type: str

creds¶

A dictionary of creds to the associated Creds object containing its credentials.

Type: dict[str, Creds]

default_alias :str¶

provided_alias :str¶

credentials :Dict[str, Credentials]¶

defaults :Dict¶

property creds(self)¶: Credentials alias used by current Connection.

property current(self)¶: Returns current credentials.

property connect_kwargs(self) → Dict¶: Arguments from snowmobile.toml for snowflake.connector.connect().

class snowmobile.core.cfg.Credentials¶

Bases: snowmobile.core.cfg.base.Base

[connection.credentials.credentials_alias]

user :str¶

password :str¶

role :str¶

account :str¶

warehouse :str¶

database :str¶

schema_name :str¶

as_nm(self, n: str)¶: Sets the credentials alias.

property credentials(self)¶: Returns namespace as a dictionary, excluding _alias.

class snowmobile.core.cfg.Loading¶

Bases: snowmobile.core.cfg.base.Base

[loading]

Default settings to use when loading data

default-file-format

Name of file-format to use when loading data into the warehouse; default is snowmobile_default_csv; which will be created and dropped afterwards if an existing file format is not specified;

Type: str

include_index¶

Include the index of a DataFrame when loading it into a table; default is False.

Type: bool

on_error¶

Action to take if an error is encountered when loading data into the warehouse; default is continue.

Type: bool

keep_local¶

Option to keep the local file exported when loading into a staging table; default is False.

Type: bool

include_loaded_tmstmp¶

Include a loaded_tmstmp column when loading a DataFrame into the warehouse; default is True.

Type: bool

quote_char¶

Quote character to use for delimited files; default is double quotes (").

Type: str

auto_compress¶

Auto-compress file when loading data; default is True..

Type: bool

overwrite_pre_existing_stage¶

Overwrite pre-existing staging table if data is being appended into an existing table/the staging table already exists; default is True.

Type: bool

defaults :Dict¶

put :Put¶

copy_into :Copy¶

export_options :Dict[str, Dict]¶

property configured_args(self) → Dict¶: Placeholder for configuration arguments of derived classes.

class snowmobile.core.cfg.Put¶

Bases: snowmobile.core.cfg.base.Base

[loading.put]

auto_compress :bool¶

class snowmobile.core.cfg.Copy¶

Bases: snowmobile.core.cfg.base.Base

[loading.copy]

on_error :str¶

class snowmobile.core.cfg.SQL¶

Bases: snowmobile.core.cfg.base.Base

[sql] (snowmobile-ext.toml)

generic_anchors :Dict¶

kw_exceptions :Dict¶

named_objects :List¶

info_schema_exceptions :Dict[str, str]¶

desc_is_simple :bool¶

pr_over_ge :bool¶

info_schema_loc(self, obj: str, stem: bool = False) → str ¶

Returns information schema table for object if other than making plural.

i.e.:

‘tables’ -> ‘tables’
‘table’ –> ‘tables’
‘schemas’ -> ‘schemata’
‘schema’ –> ‘schemata’

objects_within(self, first_line: str)¶: Searches the first line of sql for matches to named objects.

class snowmobile.core.cfg.Location¶

Bases: snowmobile.core.cfg.base.Base

[external-sources]

ddl :Path¶

extensions :Path¶

sql_export_heading :Path¶

class snowmobile.core.cfg.QA¶

Bases: snowmobile.core.cfg.base.Base

[script.qa]

partition_on :str¶

ignore_patterns :List¶

compare_patterns :List¶

tolerance :Tolerance¶

class snowmobile.core.cfg.Attributes(**data)¶

Bases: snowmobile.core.cfg.base.Base

[script.markdown.attributes]

excluded :List[str]¶

from_namespace :Dict[str, str]¶

groups :Dict¶

order :List[str]¶

reserved :Dict[str, Reserved]¶

markers :Dict[str, Marker]¶

exclude(self, item: str)¶: Adds an item (argument name) to list of exclusions.

get_marker(self, name: str)¶: Fetches a template marker from markers.

merge_markers(self, parsed_markers: Dict[int, Dict]) → Dict[int, Marker]¶

Merges markers parsed from script with template markers in snowmobile.toml.

Does the following:

Consumes all parsed attributes from markers found in a script
Tries to pull a pre-configured marker based on its time and updates its pre-configured values with those provided in the script if so
If it doesn’t find a pre-configured marker based on the marker name, it will instantiate a new marker instance from the arguments provided in the script.

Parameters: parsed_markers (Dict[int, Dict]) – All parsed raw marker arguments from a script by index position.

Returns (Dict[int, Marker]):: Instantiated markers for all attributes, merging those with matching names to pre-configured markers in snowmobile.toml.

get_position(self, attr: str) → int ¶

Returns the position for an attribute based on snowmobile-ext.toml.

Will return 0 if not included in order-by configuration.

included(self, attrs: Dict) → Dict¶: Checks if an attribute has been marked for exclusion from render.

group_parsed_attrs(self, parsed: Dict) → Dict¶: Nests attributes into dictionaries that are configured as groups.

add_reserved_attrs(self, attrs: Dict, is_marker: bool = False)¶: Batch modifies all reserved attributes to their configuration.

class snowmobile.core.cfg.Markup¶

Bases: snowmobile.core.cfg.base.Base

[script.markup]

hx_marker :str¶

hx_statement :str¶

bullet_char :str¶

attr_nm_wrap_char :str¶

attr_value_wrap_char :str¶

attrs :Attributes¶

pref_header(self, is_marker: bool = False, from_wc: Optional[str] = False) → str ¶: Creates header prefix based on specifications.

class snowmobile.core.cfg.Marker(**data)¶

Bases: snowmobile.core.cfg.base.Base

[script.markdown.attributes.markers]

name :str¶

group :str¶

attrs :Dict¶

raw :str¶

index :int¶

add(self, attrs: Dict) → snowmobile.core.cfg.script.Marker ¶: Add to existing attributes.

split_attrs(self, attrs: Dict) → Tuple[Dict, Dict]¶

Splits attributes into user-defined-only and shared with snowmobile.toml.

Parameters: attrs (Dict) – Dictionary of parsed arguments.

Returns (Tuple[Dict, Dict]):: Tuple of (shared_with_snowmobile_toml_attrs, new_attrs)

update(self, attrs: Dict) → snowmobile.core.cfg.script.Marker ¶: Merges parsed attributes with configuration attributes

set_name(self, name: str, overwrite: bool = False) → snowmobile.core.cfg.script.Marker ¶: Sets the name attribute.

as_args(self)¶: Returns a dictionary of arguments for Section.

nm(self)¶: Marker name.

class snowmobile.core.cfg.Pattern¶

Bases: snowmobile.core.cfg.base.Base

[script.patterns]

core :Core¶

wildcards :Wildcard¶

class snowmobile.core.cfg.Script¶

Bases: snowmobile.core.cfg.base.Base

[script]

patterns :Pattern¶

markup :Markup¶

qa :QA¶

types :Type¶

export_dir_nm :str¶

result_limit :int¶

tag(self) → Tuple[str, str]¶: Open/close pattern for statement tags.

static power_strip(val_to_strip: str, chars_to_strip: Iterable[str]) → str ¶: Exhaustively strips a string of specified leading/trailing characters.

arg_to_string(self, arg_as_str: str) → str ¶: Strips an argument as a string down to its elemental form.

arg_to_list(self, arg_as_str: str) → List[str]¶: Converts a list as a string into a list.

arg_to_float(self, arg_as_str: str) → float ¶: Strips a string down to its elemental form and converts to a float.

arg_to_bool(self, arg_as_str: str) → bool ¶: Converts a boolean in string-form into a boolean value.

parse_arg(self, arg_key: str, arg_value: str)¶: Parses an argument into its target data type based on its arg_key and the script.name-to-type-xref defined in snowmobile.toml.

static split_args(args_str: str) → List[str]¶: Returns a list of arguments based on splitting string on double underscores and stripping results.

parse_split_arguments(self, splitter: List[str]) → Dict¶: Returns a dictionary of argument-index to argument keys and values.

parse_str(self, block: str, strip_blanks: bool = False, strip_trailing: bool = False) → Dict¶

Parses a string of statement tags/arguments into a valid dictionary.

Parameters

block (str) – Raw string of all text found between a given open/close wrap.
strip_blanks (bool) – Strip blank lines from string; defaults to False.
strip_trailing (bool) – Strip trailing whitespace from the string; defaults to False.

Returns (dict):: Dictionary of arguments.

wrap(self, tag: str) → str ¶: Wraps a raw string of sql in open/closing patterns.

static attr_construct(attr_nm: str, attr_value: str) → str ¶: Returns a parsable attribute from an attribute name and value.

tag_from_attrs(self, attrs: Dict, nm: Optional[str] = None, wrap: bool = False, **kwargs) → str ¶: Construct a parsable tag out of a dictionary of attributes.

as_parsable(self, raw: str, is_multiline: Optional[bool] = None, is_marker: Optional[bool] = None, lines: Optional[int] = None) → str ¶

Returns a raw string wrapped in open/close tags.

Used for taking a raw string of marker or statement attributes and wrapping it in open/close tags before exporting, making the script being exported re-parsable by snowmobile.

find_spans(self, sql: str) → Dict[int, Tuple[int, int]]¶: Finds indices of script tags given a sql script as a string and an open and close pattern of the tags.

find_tags(self, sql: str) → Dict[int, str]¶: Finds indices of script tags given a sql script as a string and an open and close pattern of the tags.

find_block(self, sql: str, marker: str) → str ¶

Finds a block of arguments based on a marker.

Markers expected by default are the __script__ and __appendix__ markers.

has_tag(self, s: sqlparse.sql.Statement) → bool ¶: Checks if a given statement has a wrap that directly precedes the sql.

static is_marker(raw: str)¶: Checks if a raw string of arguments has a marker on the first line.

static is_valid_sql(s: sqlparse.sql.Statement) → bool ¶: Verifies that a given sqlparse.sql.Statement contains valid sql.

static strip_comments(s: sqlparse.sql.Statement) → str ¶: Isolates just the sql within a sqlparse.sql.Statement object.

split_sub_blocks(self, s: sqlparse.sql.Statement) → Tuple[List, str]¶

Breaks apart blocks of arguments within a sqlparse.sql.Statement.

Note

sqlparse.parsestream() returns a list of sqlparse.sql.Statement objects, each of which includes all text (comments) between the last character of the prior statement and the first character of the current one.
split_sub_blocks() traverses that space and identifies all spans of text wrapped in open (/*-) and close (-*/) tags, storing their index positions relative to the other statements & markers.
These are stored as snowmobile.core.Script attributes as statements are parsed and so that they can be exported in the appropriate order to a markdown file.

Parameters: s (sqlparse.sql.Statement) – sqlparse.sql.Statement object.

Returns (Tuple[List, str]):

A tuple containing:

A list of __marker__ blocks if they exist; empty list otherwise
The last wrap/block before the start of the actual SQL (e.g. the wrap/block that is associated with the statement passed to s.

name_from_marker(self, raw: str) → str ¶: Extracts a marker name (e.g. ‘script’ from within __script__).

parse_name(self, raw: str, offset: Optional[int] = None, silence: bool = False) → str ¶: Parses name from a raw set of arguments if not given an explicit wrap.

static add_name(nm_title: str, nm_marker: str, attrs: dict, overwrite: bool = False)¶

Adds a name to a set of parsed marker attributes.

Accepts a name and a dict of parsed attributes from a marker and:

Checks to see if there’s an explicit ‘name’ declared within the attributes
If not explicitely declared or explicitely declared and overwrite=False, it will add the nm value to the attributes as ‘name’.
It will also add the ‘nm’ value to the attributes as ‘marker-name’ to be used by the Marker when cross-referencing the __name__ with template markers in snowmobile.toml.

Parameters

nm_title (str) –
The name of the marker as either:
1. Returned value from name_from_marker()
2. Returned value from parse_name()
3. None if neither is provided nm_marker (str):
The string value wrapped in __ on the first line of the argument block.
attrs (dict) – A dictionary of parsed attributes as returned from parse_str().
overwrite (bool) – Indicator of whether or not to overwrite a ‘name’ attribute declared within the .sql script.

parse_marker(self, attrs_raw: str) → Dict¶: Parses a raw string of __marker__ text between an open and a close pattern.

static ensure_sqlparse(sql: Union[sqlparse.sql.Statement, str]) → sqlparse.sql.Statement ¶

Returns a sqlparse.sql.Statement from sql.

Will return sql with no modification if it’s already a sqlparse.sql object.

Needed to accommodate dynamic addition of statements as strings to an existing Script object from from raw strings as opposed to a sqlparse.sql.Statement objects as is done when reading a sql file.

Parameters: sql (Union[sqlparse.sql.Statement, str]) – Either a string of sql or an already parsed sqlparse.sql.Statement object.

Returns (sqlparse.sql.Statement):: A parsed sql statement.

sql_tokens(self, sql: str) → List[sqlparse.sql.Token]¶

Unpacks nested tokens from a sqlparse.sql.Statement.

Parameters: sql (str) – A raw sql from a statement.
Returns: A list of tokens.

id_from_tokens(self, sql: str) → str ¶

Identifies the last identifier in a piece of raw sql.

Identifies obj being operated on.

Parameters: sql (str) – A raw piece of sql from a statement.
Returns: A string if the last identifier found or an empty string otherwise.

class snowmobile.core.cfg.Wildcard¶

Bases: snowmobile.core.cfg.base.Base

[script.patterns.wildcards]

char_wc :str¶

char_sep :str¶

wc_paragraph :str¶

wc_as_is :str¶

wc_omit_attr_nm :str¶

find_first_wc_idx(self, attr_nm: str) → int ¶

Finds index of the first unescaped wildcard in an attribute name.

Parameters: attr_nm (str) – Attribute name to search through.

Returns (int):: Index position of first unescaped wildcard or 0 if one does not exist.

partition_on_wc(self, attr_nm: str) → Tuple[str, List[str]]¶

Parses attribute name into its display name and its wildcards.

Uses Wildcard.find_first_wc_idx() to determine if attr_nm contains a valid wildcard.

Parameters: attr_nm (str) – Attribute name to parse.

Returns (Tuple[str, List[str]]):: Tuple containing the attribute display name and a list of its wildcards if present and an empty list otherwise.

snowmobile.core.cfg¶

Submodules¶

Package Contents¶

Classes¶

`snowmobile.core.cfg`¶