Parses table column names and types from specified SQLite table schema >>> kb.data.cachedColumns = {} >>> parseSqliteTableSchema("CREATE TABLE users(\\n\\t\\tid INTEGER,\\n\\t\\tname TEXT\\n);") True >>> tuple(kb.data.cachedColumns[conf.db][conf.tbl].items()) == (('id', 'INTEGE
(value)
| 3420 | return retVal |
| 3421 | |
| 3422 | def parseSqliteTableSchema(value): |
| 3423 | """ |
| 3424 | Parses table column names and types from specified SQLite table schema |
| 3425 | |
| 3426 | >>> kb.data.cachedColumns = {} |
| 3427 | >>> parseSqliteTableSchema("CREATE TABLE users(\\n\\t\\tid INTEGER,\\n\\t\\tname TEXT\\n);") |
| 3428 | True |
| 3429 | >>> tuple(kb.data.cachedColumns[conf.db][conf.tbl].items()) == (('id', 'INTEGER'), ('name', 'TEXT')) |
| 3430 | True |
| 3431 | >>> parseSqliteTableSchema("CREATE TABLE dummy(`foo bar` BIGINT, \\"foo\\" VARCHAR, 'bar' TEXT)"); |
| 3432 | True |
| 3433 | >>> tuple(kb.data.cachedColumns[conf.db][conf.tbl].items()) == (('foo bar', 'BIGINT'), ('foo', 'VARCHAR'), ('bar', 'TEXT')) |
| 3434 | True |
| 3435 | >>> parseSqliteTableSchema("CREATE TABLE suppliers(\\n\\tsupplier_id INTEGER PRIMARY KEY DESC,\\n\\tname TEXT NOT NULL\\n);"); |
| 3436 | True |
| 3437 | >>> tuple(kb.data.cachedColumns[conf.db][conf.tbl].items()) == (('supplier_id', 'INTEGER'), ('name', 'TEXT')) |
| 3438 | True |
| 3439 | >>> parseSqliteTableSchema("CREATE TABLE country_languages (\\n\\tcountry_id INTEGER NOT NULL,\\n\\tlanguage_id INTEGER NOT NULL,\\n\\tPRIMARY KEY (country_id, language_id),\\n\\tFOREIGN KEY (country_id) REFERENCES countries (country_id) ON DELETE CASCADE ON UPDATE NO ACTION,\\tFOREIGN KEY (language_id) REFERENCES languages (language_id) ON DELETE CASCADE ON UPDATE NO ACTION);"); |
| 3440 | True |
| 3441 | >>> tuple(kb.data.cachedColumns[conf.db][conf.tbl].items()) == (('country_id', 'INTEGER'), ('language_id', 'INTEGER')) |
| 3442 | True |
| 3443 | """ |
| 3444 | |
| 3445 | retVal = False |
| 3446 | |
| 3447 | value = extractRegexResult(r"(?s)\((?P<result>.+)\)", value) |
| 3448 | |
| 3449 | if value: |
| 3450 | table = {} |
| 3451 | columns = OrderedDict() |
| 3452 | |
| 3453 | value = re.sub(r"\(.+?\)", "", value).strip() |
| 3454 | |
| 3455 | for match in re.finditer(r"(?:\A|,)\s*(([\"'`]).+?\2|\w+)(?:\s+(INT|INTEGER|TINYINT|SMALLINT|MEDIUMINT|BIGINT|UNSIGNED BIG INT|INT2|INT8|INTEGER|CHARACTER|VARCHAR|VARYING CHARACTER|NCHAR|NATIVE CHARACTER|NVARCHAR|TEXT|CLOB|LONGTEXT|BLOB|NONE|REAL|DOUBLE|DOUBLE PRECISION|FLOAT|REAL|NUMERIC|DECIMAL|BOOLEAN|DATE|DATETIME|NUMERIC)\b)?", decodeStringEscape(value), re.I): |
| 3456 | column = match.group(1).strip(match.group(2) or "") |
| 3457 | if re.search(r"(?i)\A(CONSTRAINT|PRIMARY|UNIQUE|CHECK|FOREIGN)\b", column.strip()): |
| 3458 | continue |
| 3459 | retVal = True |
| 3460 | |
| 3461 | columns[column] = match.group(3) or "TEXT" |
| 3462 | |
| 3463 | table[safeSQLIdentificatorNaming(conf.tbl, True)] = columns |
| 3464 | kb.data.cachedColumns[conf.db] = table |
| 3465 | |
| 3466 | return retVal |
| 3467 | |
| 3468 | def getTechniqueData(technique=None): |
| 3469 | """ |
no test coverage detected
searching dependent graphs…