Changeset 4425

Timestamp:

04/04/09 15:26:09 (4 years ago)

Author:

gpolo

Message:

Added function for extracting column-defs from create table statement.

Location:

branch/merger

Files:

• sqlitedb.py (modified) (3 diffs)
• tests/test_sqlitemerge.py (modified) (3 diffs)

branch/merger/sqlitedb.py

@@ -2 +2 @@
 
 import os
+import re
 import shutil
 import warnings
@@ -24 +25 @@
 #   Index merging (read copying)
 #   Proper foreign key merging
+VALID_NAME_PATTERN = (
+        """
+        (
+            # If name starts with [ or ` or " then I assume anything
+            # can form the name. Note that is not really important to
+            # know if pairs of [ ] (or " ", ` `) matches, it is assumed
+            # that sqltable was validated by sqlite already.
+            ((\[ | ` | ") .+ (\] | ` | ")) |
+            # Otherwise just [a-zA-Z0-9_] (maybe more depending on the flags).
+            (\w+)
+        )
+        """)
+VALID_NAME = re.compile(VALID_NAME_PATTERN, re.VERBOSE)
+
+def column_definitions(sqltable):
+    """Given a complete create-table-stmt defined in sqltable, return the
+    columns on it and their respectives column-defs."""
+    sqltable = sqltable.strip() or ''
+
+    # Discarding all "CREATE [TEMP|TEMPORARY] TABLE [IF NOT EXISTS]"
+    discard = re.match(
+            """
+            ^\s*CREATE\s*
+            (TEMP|TEMPORARY)?\s*
+            TABLE\s*
+            (IF\s*NOT\s*EXISTS\s*)?\s+
+            """, sqltable, re.IGNORECASE | re.VERBOSE)
+    if discard is None:
+        raise Exception("Malformed create-table-stmt")
+
+    sqltable = sqltable[discard.end():]
+
+    # Discarding [dbname].tablename
+    discard = re.match(
+            """
+            # optional dbname
+            ^(.+\.)?
+            # Looking for tablename now.
+            %s
+            # Spaces
+            \s*
+            """ % VALID_NAME_PATTERN, sqltable, re.VERBOSE)
+    if discard is None:
+        raise Exception("Table name not present")
+
+    if sqltable[:2].lower() == 'as':
+        raise Exception("Table will be formed by the result set of "
+                "a query, not supported.")
+
+    if sqltable[discard.end()] != '(' or sqltable[-1] != ')':
+        raise Exception("Malformed create-table-stmt, missing parentheses")
+
+    interesting = sqltable[discard.end()+1:-1].lstrip()
+
+    # Now we have column-def, column-def, .. [, table-constraint ..] and
+    # we want to split each column-def into
+    # (column-name, type-name [column-constraint])
+    column_defs = {}
+
+    table_constraints = set([
+        'constraint', 'primary', 'unique', 'check', 'foreign'])
+
+    while interesting:
+        column_name = VALID_NAME.match(interesting)
+        if column_name is None:
+            raise Exception("Missing column-name")
+        interesting = interesting[column_name.end():].lstrip()
+        column_name = column_name.group().lower()
+        if column_name in table_constraints:
+            # We are not interested in table-contrainst.
+            # XXX This can be used to indicate that this table can't be fully
+            # merged.
+            continue
+
+        # Now there is either a type-name, column-constraint, or nothing.
+
+        # We want to either find the next comma that separates this
+        # column-def to the next one, or the end of input meaning this
+        # was the last column-def.
+        in_parens = False
+        for indx, char in enumerate(interesting):
+            if char == ',' and not in_parens:
+                # Found the comma we were after.
+                column_defs[column_name] = interesting[:indx]
+                break
+            elif char == '(':
+                in_parens = True
+            elif char == ')':
+                in_parens = False
+        else:
+            # Comma not found, column-def finished.
+            column_defs[column_name] = interesting or None
+            break
+
+        interesting = interesting[indx + 1:].lstrip()
+
+    return column_defs
+
 
 class SqliteDBMerge(object):
@@ -200 +299 @@
             self._tocursor.execute(result['sql'])
 
+
 merge = SqliteDBMerge
 

branch/merger/tests/test_sqlitemerge.py

@@ -3 +3 @@
 from test.test_support import run_unittest
 
-from sqlitedb import sqlite, merge, _DictCursor
+from sqlitedb import sqlite, merge, column_definitions, _DictCursor
 
 def conn_cursor(name):
@@ -40 +40 @@
     def merge_now(self):
         merge(self.new_db_file.name, self.old_db_file.name)
+
+
+class ColumnDefinitionTest(unittest.TestCase):
+
+    def test_invalid(self):
+        self.failUnlessRaises(Exception, column_definitions, 'buh')
+        self.failUnlessRaises(Exception, column_definitions, None)
+
+    def test_simple_createtable(self):
+        ctable_stmt = "CREATE TABLE x (a UNIQUE)"
+        self.failUnlessEqual(column_definitions(ctable_stmt),
+                {'a': 'UNIQUE'})
+        ctable_stmt = (
+                "CREATE TABLE x ("
+                "   a UNIQUE CHECK (1, 2), "
+                "   b CONSTRAINT buh)")
+        self.failUnlessEqual(column_definitions(ctable_stmt),
+                {'a': 'UNIQUE CHECK (1, 2)', 'b': 'CONSTRAINT buh'})
 
 
@@ -154 +172 @@
 
 if __name__ == "__main__":
-    run_unittest(ColumnMergeTest, TriggersTest, TableCopyTest)
+    run_unittest(ColumnDefinitionTest, ColumnMergeTest, TriggersTest,
+            TableCopyTest)