Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
},
},
"datatype_postgres": {
"source_ddl":
"""
CREATE TABLE Test_Table (
Col_01 character varying(100) PRIMARY KEY,
Col_02 json NOT NULL,
Col_03 integer[], -- one dimensional array
Col_04 integer[][], -- two dimensional array
Col_05 integer[][][], -- multiple dimensional array
Col_06 character varying[] -- character varying array
);
""",
"bq_ddl": {
DdlParse.NAME_CASE.original:
"""\
#standardSQL
CREATE TABLE `project.dataset.Test_Table`
(
Col_01 STRING NOT NULL,
Col_02 STRING NOT NULL,
Col_03 ARRAY,
Col_04 ARRAY>>,
Col_05 ARRAY>>>>,
Col_06 ARRAY
)""",
DdlParse.NAME_CASE.lower:
"""\
#standardSQL
CREATE TABLE `project.dataset.test_table`
(
"ddl":
"""
CREATE TABLE Sample_Table (
Col_01 varchar(100),
Col_02 char(200),
Col_03 integer,
Col_04 double,
Col_05 datetime,
Col_06 decimal(2,1) not null default '0.0',
Col_07 integer,
CONSTRAINT const_01 PRIMARY KEY (Col_01, Col_02),
CONSTRAINT \"const_02\" UNIQUE (Col_03, Col_04),
CONSTRAINT \"const_03\" FOREIGN KEY (Col_04, \"Col_05\") REFERENCES ref_table_01 (\"Col_04\", Col_05)
);
""",
"database": DdlParse.DATABASE.mysql,
"table": {"schema": None, "name": "Sample_Table", "temp": False},
"columns": [
{"name": "Col_01", "type": "VARCHAR", "length": 100, "scale": None, "array_dimensional": 0, "is_unsigned": False, "is_zerofill": False, "not_null": True, "pk": True, "unique": False, "constraint": "PRIMARY KEY", "description": None},
{"name": "Col_02", "type": "CHAR", "length": 200, "scale": None, "array_dimensional": 0, "is_unsigned": False, "is_zerofill": False, "not_null": True, "pk": True, "unique": False, "constraint": "PRIMARY KEY", "description": None},
{"name": "Col_03", "type": "INTEGER", "length": None, "scale": None, "array_dimensional": 0, "is_unsigned": False, "is_zerofill": False, "not_null": False, "pk": False, "unique": True, "constraint": "UNIQUE", "description": None},
{"name": "Col_04", "type": "DOUBLE", "length": None, "scale": None, "array_dimensional": 0, "is_unsigned": False, "is_zerofill": False, "not_null": False, "pk": False, "unique": True, "constraint": "UNIQUE", "description": None},
{"name": "Col_05", "type": "DATETIME", "length": None, "scale": None, "array_dimensional": 0, "is_unsigned": False, "is_zerofill": False, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_06", "type": "DECIMAL", "length": 2, "scale": 1, "array_dimensional": 0, "is_unsigned": False, "is_zerofill": False, "not_null": True, "pk": False, "unique": False, "constraint": "NOT NULL", "description": None},
{"name": "Col_07", "type": "INTEGER", "length": None, "scale": None, "array_dimensional": 0, "is_unsigned": False, "is_zerofill": False, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
],
"bq_field": [
'{"name": "Col_01", "type": "STRING", "mode": "REQUIRED"}',
'{"name": "Col_02", "type": "STRING", "mode": "REQUIRED"}',
'{"name": "Col_03", "type": "INTEGER", "mode": "NULLABLE"}',
'{"name": "Col_04", "type": "FLOAT", "mode": "NULLABLE"}',
'{"name": "Col_05", "type": "DATETIME", "mode": "NULLABLE"}',
def test_parse(test_case, parse_pattern):
# Get test ddl script
data = TEST_DATA[test_case]
# Create parse object instance
ddlparse = DdlParse()
# Set source database option
# Set source database option & Parse ddl
if parse_pattern == DDL_SET_PATTERN.method:
if data["database"] is not None:
table = ddlparse.parse(data["ddl"], data["database"])
else:
table = ddlparse.parse(data["ddl"])
else:
if data["database"] is not None:
ddlparse.source_database = data["database"]
ddlparse.ddl = data["ddl"]
table = ddlparse.parse()
def test_set_comment():
ddl = """
CREATE TABLE Sample_Table (
Col_01 integer
);
"""
# Parse DDL
table = DdlParse().parse(ddl)
# Check blank comment
assert table.columns["Col_01"].comment is None
assert table.columns["Col_01"].description is None
# Check comment setter
table.columns["Col_01"].comment = "comment_1"
assert table.columns["Col_01"].comment == "comment_1"
# Check description setter
table.columns["Col_01"].description = "comment_2"
assert table.columns["Col_01"].description == "comment_2"
def test_exception_ddl():
# Create parse object instance
ddlparse = DdlParse()
# Do not set DDL
# Error : DDL is not specified
with pytest.raises(ValueError):
ddlparse.parse()
def test_exception_bq_data_type():
ddl = """
CREATE TABLE Sample_Table (
Col_01 NG_DATA_TYPE,
);
"""
# Parse DDL
table = DdlParse().parse(ddl)
# Error: Unknown data type
with pytest.raises(ValueError):
print(table.columns["col_01"].bigquery_data_type)
def test_bq_ddl(test_case):
# Get test data
data = TEST_DATA_DDL[test_case]
# Parse ddl
table = DdlParse().parse(data["source_ddl"])
# Check generate BigQuery DDL statements of DdlParseTable
assert table.to_bigquery_ddl() == textwrap.dedent(data["bq_ddl"][DdlParse.NAME_CASE.original])
assert table.to_bigquery_ddl(DdlParse.NAME_CASE.original) == textwrap.dedent(data["bq_ddl"][DdlParse.NAME_CASE.original])
assert table.to_bigquery_ddl(DdlParse.NAME_CASE.lower) == textwrap.dedent(data["bq_ddl"][DdlParse.NAME_CASE.lower])
assert table.to_bigquery_ddl(DdlParse.NAME_CASE.upper) == textwrap.dedent(data["bq_ddl"][DdlParse.NAME_CASE.upper])
Col_403 MEDIUMINT(13) UNSIGNED ZEROFILL NOT NULL,
Col_404 INT(14) UNSIGNED ZEROFILL NOT NULL,
Col_405 INTEGER(15) UNSIGNED ZEROFILL NOT NULL,
Col_406 BIGINT(16) UNSIGNED ZEROFILL NOT NULL,
Col_407 DECIMAL(17, 1) UNSIGNED ZEROFILL NOT NULL,
Col_408 DEC(18, 2) UNSIGNED ZEROFILL NOT NULL,
Col_409 NUMERIC(19, 3) UNSIGNED ZEROFILL NOT NULL,
Col_410 FIXED(20, 4) UNSIGNED ZEROFILL NOT NULL,
Col_411 FLOAT(21, 5) UNSIGNED ZEROFILL NOT NULL,
Col_412 DOUBLE(22, 6) UNSIGNED ZEROFILL NOT NULL,
Col_413 DOUBLE PRECISION(23, 7) UNSIGNED ZEROFILL NOT NULL,
Col_414 REAL(24, 8) UNSIGNED ZEROFILL NOT NULL,
Col_415 FLOAT(25) UNSIGNED ZEROFILL NOT NULL
);
""",
"database": DdlParse.DATABASE.mysql,
"table": {"schema": None, "name": "Sample_Table", "temp": False},
"columns": [
{"name": "Col_101", "type": "TINYINT", "length": 11, "scale": None, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_102", "type": "SMALLINT", "length": 12, "scale": None, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_103", "type": "MEDIUMINT", "length": 13, "scale": None, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_104", "type": "INT", "length": 14, "scale": None, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_105", "type": "INTEGER", "length": 15, "scale": None, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_106", "type": "BIGINT", "length": 16, "scale": None, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_107", "type": "DECIMAL", "length": 17, "scale": 1, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_108", "type": "DEC", "length": 18, "scale": 2, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_109", "type": "NUMERIC", "length": 19, "scale": 3, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_110", "type": "FIXED", "length": 20, "scale": 4, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_111", "type": "FLOAT", "length": 21, "scale": 5, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_112", "type": "DOUBLE", "length": 22, "scale": 6, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_113", "type": "DOUBLE PRECISION", "length": 23, "scale": 7, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
{"name": "Col_114", "type": "REAL", "length": 24, "scale": 8, "array_dimensional": 0, "is_unsigned": True, "is_zerofill": True, "not_null": False, "pk": False, "unique": False, "constraint": "", "description": None},
TEST_DATA_DDL = {
"exist_schema_name": {
"source_ddl":
"""
CREATE TABLE Test_Schema.Test_Table (
Col_01 varchar(100) PRIMARY KEY,
Col_02 char(200) NOT NULL UNIQUE,
Col_03 integer UNIQUE,
Col_04 double,
Col_05 datetime NOT NULL,
Col_06 bool
);
""",
"bq_ddl": {
DdlParse.NAME_CASE.original:
"""\
#standardSQL
CREATE TABLE `project.Test_Schema.Test_Table`
(
Col_01 STRING NOT NULL,
Col_02 STRING NOT NULL,
Col_03 INT64,
Col_04 FLOAT64,
Col_05 DATETIME NOT NULL,
Col_06 BOOL
)""",
DdlParse.NAME_CASE.lower:
"""\
#standardSQL
CREATE TABLE `project.test_schema.test_table`
(
""",
"bq_ddl": {
DdlParse.NAME_CASE.original:
"""\
#standardSQL
CREATE TABLE `project.dataset.Test_Table`
(
Col_01 STRING NOT NULL OPTIONS (description = "Single Quote"),
Col_02 STRING NOT NULL OPTIONS (description = "Double Quote"),
Col_03 ARRAY OPTIONS (description = "in \\"Quote\\""),
Col_04 ARRAY>> OPTIONS (description = "in 'Quote'"),
Col_05 ARRAY>>>> OPTIONS (description = "コメント is full-width(Japanese) character"),
Col_06 STRING OPTIONS (description = "Comma, strings, ,"),
Col_07 ARRAY
)""",
DdlParse.NAME_CASE.lower:
"""\
#standardSQL
CREATE TABLE `project.dataset.test_table`
(
col_01 STRING NOT NULL OPTIONS (description = "Single Quote"),
col_02 STRING NOT NULL OPTIONS (description = "Double Quote"),
col_03 ARRAY OPTIONS (description = "in \\"Quote\\""),
col_04 ARRAY>> OPTIONS (description = "in 'Quote'"),
col_05 ARRAY>>>> OPTIONS (description = "コメント is full-width(Japanese) character"),
col_06 STRING OPTIONS (description = "Comma, strings, ,"),
col_07 ARRAY
)""",
DdlParse.NAME_CASE.upper:
"""\
#standardSQL
CREATE TABLE `project.dataset.TEST_TABLE`