introduce pyparsing

.isort.cfg

@@ -1,2 +1,2 @@
 [settings]
-known_third_party = black,click,jinja2,pytest,setuptools,sql_to_code
+known_third_party = black,click,jinja2,pyparsing,pytest,setuptools,sql_to_code

sql_to_code/parsers/alter_table/parser.py

@@ -1,32 +1,33 @@
-import re
+from pyparsing import *
 
 from .models import ForeignKey, Reference
 
-source_table_name_regex = re.compile('ALTER TABLE "(?P<table_name>\w+)"')
-foreign_key_name_regex = re.compile('ADD FOREIGN KEY \("(?P<foreign_key>\w+)"\)')
-result_table_name_regex = re.compile('REFERENCES "(?P<result_table_name>\w+)"')
-result_table_field_name_regex = re.compile(
-    'REFERENCES "\w+" \("(?P<result_table_field_name>\w+)"\);'
+table_name_schema = CaselessKeyword("alter table") + QuotedString('"')("table_name")
+foreign_key_schema = CaselessKeyword("add foreign key") + QuotedString(
+    '("', endQuoteChar='")'
+)("foreign_key")
+reference_table = CaselessKeyword("references") + QuotedString('"')("reference_table")
+reference_table_column_name = QuotedString('("', endQuoteChar='")')(
+    "reference_table_column_name"
+)
+
+add_foreign_key_schema = (
+    table_name_schema
+    + foreign_key_schema
+    + reference_table
+    + reference_table_column_name
 )
 
 
 def parse(sql_text: str):
-    source_table_name = source_table_name_regex.search(sql_text).groupdict()[
-        "table_name"
-    ]
-    foreign_key_name = foreign_key_name_regex.search(sql_text).groupdict()[
-        "foreign_key"
-    ]
-    result_table_name = result_table_name_regex.search(sql_text).groupdict()[
-        "result_table_name"
-    ]
-    result_table_field_name = result_table_field_name_regex.search(
-        sql_text
-    ).groupdict()["result_table_field_name"]
+    result = add_foreign_key_schema.parseString(sql_text)
 
     return ForeignKey(
-        refer_from=Reference(table_name=source_table_name, field_name=foreign_key_name),
+        refer_from=Reference(
+            table_name=result.table_name, field_name=result.foreign_key
+        ),
         refer_to=Reference(
-            table_name=result_table_name, field_name=result_table_field_name
+            table_name=result.reference_table,
+            field_name=result.reference_table_column_name,
         ),
     )

sql_to_code/parsers/create_enum/parser.py

@@ -1,13 +1,18 @@
-import re
+from pyparsing import *
 
 from .models import Enumeration
 
-NAME_REGEX = re.compile('CREATE TYPE "(?P<name>\w+)"')
-VALUE_REGEX = re.compile("'(\w+)'")
+enum_schema = (
+    CaselessKeyword("create type")
+    + QuotedString('"')("enum_name")
+    + CaselessKeyword("as enum")
+    + CaselessKeyword("(")
+    + Group(OneOrMore(QuotedString("'") + Optional(Suppress(","))))("enum_values")
+    + CaselessKeyword(");")
+)
 
 
 def parse(sql_text: str):
-    name = NAME_REGEX.search(sql_text).groupdict()["name"]
-    values = VALUE_REGEX.findall(sql_text)
+    result = enum_schema.parseString(sql_text)
 
-    return Enumeration(name, values)
+    return Enumeration(result.enum_name, result.enum_values.asList())

sql_to_code/parsers/create_table/models.py

@@ -10,12 +10,20 @@
 class Attribute:
     name: str
     type: str
-    is_default: bool
     default: default_type
-    primary_key: bool
-    nullable: bool
+    is_unique: bool
+    is_nullable: bool
+    is_primary_key: bool
     foreign_key: Reference = field(default=None)
 
+    @property
+    def has_default(self):
+        return self.default is not None
+
+    @property
+    def has_foreign_key(self):
+        return self.foreign_key is not None
+
 
 @dataclass
 class Table:

sql_to_code/parsers/create_table/parser.py

@@ -1,46 +1,67 @@
-import re
-from typing import List
+from pyparsing import *
 
-from .models import Attribute, Table, default_type
+from .models import Attribute, Table
 
-regex_name_schema: str = r'CREATE TABLE "([\S\d]+)"\s*\(\s*(.+)\s*\);'
-name_and_type_regex = r"(^[\S\d]+)\s([\S]+)\s*"
-default_regex = r"DEFAULT\s+(.?)+\s*"
+# table schema
+create_table = CaselessKeyword("create table")
+table_name = MatchFirst(QuotedString('"'))("table_name")
 
+# field schema
+column_name = QuotedString('"')("column_name")
 
-def parse(sql_text: str) -> Table:
-    table_name, schema = re.findall(regex_name_schema, sql_text)[0]
-    all_attributes: List[str] = schema.split(",")
-    schema: List[str] = map(
-        lambda attribute: attribute.strip().replace('"', ""), all_attributes
-    )
-    attributes = parse_attributes(schema)
+# column type:
+# "(" ")" - because of varchar(40)
+# "_" - because of enums like process_type
+column_type = Word(alphas, alphanums + "(" + ")" + "_")("column_type")
+is_primary_key = CaselessKeyword("primary key")
+is_unique = CaselessKeyword("unique")("is_unique")
+
+# default:
+# "(" ")" - because of (now())
+# '"' - because of "verification"
+has_default = CaselessKeyword("default") + Word(alphanums + '"' + "(" + ")")("default")
+is_not_null = CaselessKeyword("not null")("is_not_null")
 
-    return Table(table_name, attributes)
 
+table_columns = OneOrMore(
+    Group(
+        column_name
+        + column_type
+        + (
+            Optional(is_not_null)
+            & Optional(has_default)
+            & Optional(is_unique)
+            & Optional(is_primary_key)
+        )
+        + Optional(Suppress(","))
+    )
+)("table_columns")
 
-def parse_attributes(schema) -> List[Attribute]:
-    attributes = list()
+create_table_schema = (
+    create_table
+    + table_name
+    + CaselessKeyword("(")
+    + table_columns
+    + CaselessKeyword(");")
+)
 
-    for attribute in schema:
-        name, a_type = re.findall(pattern=name_and_type_regex, string=attribute)[0]
 
-        is_pk = "PRIMARY KEY" in attribute.upper()
-        is_default = not is_pk and "DEFAULT" in attribute.upper()
+def parse(sql_text: str) -> Table:
+    result = create_table_schema.parseString(sql_text)
 
-        attributes.append(
+    table = Table(
+        result.table_name,
+        [
             Attribute(
-                name=name,
-                type=a_type,
-                primary_key=is_pk,
-                nullable="NOT NULL" not in attribute.upper() and not is_pk,
-                is_default=is_default,
-                default=parse_default(attribute) if is_default else None,
+                name=column.column_name,
+                type=column.column_type,
+                default=column.default if column.default else None,
+                is_unique=bool(column.is_unique),
+                is_nullable=not bool(column.is_not_null),
+                is_primary_key=bool(column.is_primary_key),
             )
-        )
-
-    return attributes
-
+            for column in result.table_columns
+        ],
+    )
 
-def parse_default(sql_command: str) -> default_type:
-    return re.findall(default_regex, sql_command)[0]
+    return table

sql_to_code/utils.py

@@ -11,7 +11,6 @@ def get_file_content(filename: str):
 
 
 def parse_commands(content: str) -> List[str]:
-    raw_commands = content.split("\n\n")
-    commands = [command.replace("\n", "") for command in raw_commands]
+    commands = content.split("\n\n")
 
     return commands

tests/test_sql/test_schema_alter.sql

@@ -1,3 +1,3 @@
 ALTER TABLE "issue"
-    ADD FOREIGN KEY ("process_id")
+    ADD FOREIGN KEY ("process_id_test")
     REFERENCES "process" ("process_id");

tests/test_sql/test_schema_table.sql

@@ -1,7 +1,7 @@
 CREATE TABLE "process" (
   "process_id" int PRIMARY KEY,
-  "booking_id" int,
-  "ticket_id" int,
-  "created_at" timestamp,
-  "updated_at" timestamp
+  "booking_id" int DEFAULT 10,
+  "state" process_state DEFAULT "verification" NOT NULL,
+  "created_at" timestamp DEFAULT (now()),
+  "updated_at" timestamp,
 );