Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
))
return textwrap.dedent(
"""\
#standardSQL
CREATE TABLE `project.{dataset}.{table}`
(
{colmns_define}
)""").format(
dataset=dataset,
table=self.get_name(name_case),
colmns_define=",\n ".join(cols_defs),
)
class DdlParse(DdlParseBase):
"""DDL parser"""
_LPAR, _RPAR, _COMMA, _SEMICOLON, _DOT, _DOUBLEQUOTE, _BACKQUOTE, _SPACE = map(Suppress, "(),;.\"` ")
_CREATE, _TABLE, _TEMP, _CONSTRAINT, _NOT_NULL, _PRIMARY_KEY, _UNIQUE, _UNIQUE_KEY, _FOREIGN_KEY, _REFERENCES, _KEY, _CHAR_SEMANTICS, _BYTE_SEMANTICS = \
map(CaselessKeyword, "CREATE, TABLE, TEMP, CONSTRAINT, NOT NULL, PRIMARY KEY, UNIQUE, UNIQUE KEY, FOREIGN KEY, REFERENCES, KEY, CHAR, BYTE".replace(", ", ",").split(","))
_TYPE_UNSIGNED, _TYPE_ZEROFILL = \
map(CaselessKeyword, "UNSIGNED, ZEROFILL".replace(", ", ",").split(","))
_SUPPRESS_QUOTE = _BACKQUOTE | _DOUBLEQUOTE
_COMMENT = Suppress("--" + Regex(r".+"))
_COLUMN_CONSTRAINT_BASE = r"""
(?!--)
(
(
\s*\b(?:NOT\s+)NULL?\b
fields_cur['type'] = self.bigquery_legacy_data_type if is_last else "RECORD"
fields_cur['mode'] = self.bigquery_mode if is_last else "REPEATED"
col = OrderedDict()
col['name'] = col_name
col['type'] = type
col['mode'] = mode
if self.description is not None:
col['description'] = self.description
if self.array_dimensional > 1:
col['fields'] = fields['fields']
return json.dumps(col, ensure_ascii=False)
class DdlParseColumnDict(OrderedDict, DdlParseBase):
"""
Columns dictionary collection
* Orderd dictionary
* Dict with case insensitive keys
(SQL is case insensitive)
"""
def __init__(self, source_database=None):
super().__init__()
self.source_database = source_database
def __getitem__(self, key):
return super().__getitem__(key.lower())
def __setitem__(self, key, value):
@property
def source_database(self):
"""
Source database option
:param source_database: enum DdlParse.DATABASE
"""
return self._source_database
@source_database.setter
def source_database(self, source_database):
self._source_database = source_database
class DdlParseTableColumnBase(DdlParseBase):
def __init__(self, source_database=None):
super().__init__(source_database)
self._name = ""
@property
def name(self):
"""name"""
return self._name
@name.setter
def name(self, name):
self._name = name
def get_name(self, name_case=DdlParseBase.NAME_CASE.original):
"""