Revision 6899
Added by Aaron Marcuse-Kubitza about 12 years ago
| lib/sql.py | ||
|---|---|---|
| 469 | 469 |
self.db.commit() |
| 470 | 470 |
|
| 471 | 471 |
def col_info(self, col, cacheable=True): |
| 472 |
table = sql_gen.Table('columns', 'information_schema')
|
|
| 473 |
cols = [sql_gen.Col('data_type'), sql_gen.Col('udt_name'),
|
|
| 474 |
'column_default', sql_gen.Cast('boolean',
|
|
| 475 |
sql_gen.Col('is_nullable'))]
|
|
| 472 |
module = util.root_module(db.db) |
|
| 473 |
if module == 'psycopg2': |
|
| 474 |
qual_table = sql_gen.Literal(col.table.to_str(db)) |
|
| 475 |
col_name_str = sql_gen.Literal(col.name) |
|
| 476 |
try: |
|
| 477 |
type_, type_is_array, default, nullable = row(run_query(db, '''\ |
|
| 478 |
SELECT |
|
| 479 |
format_type(COALESCE(NULLIF(typelem, 0), pg_type.oid), -1) AS type |
|
| 480 |
, typcategory = 'A' AS type_is_array |
|
| 481 |
, pg_attrdef.adsrc AS default |
|
| 482 |
, NOT pg_attribute.attnotnull AS nullable |
|
| 483 |
FROM pg_attribute |
|
| 484 |
LEFT JOIN pg_type ON pg_type.oid = atttypid |
|
| 485 |
LEFT JOIN pg_attrdef ON adrelid = attrelid AND adnum = attnum |
|
| 486 |
WHERE |
|
| 487 |
attrelid = '''+self.esc_value(qual_table)+'''::regclass |
|
| 488 |
AND attname = '''+self.esc_value(col_name_str)+''' |
|
| 489 |
''' |
|
| 490 |
, cacheable=True, log_level=4)) |
|
| 491 |
except StopIteration: raise sql_gen.NoUnderlyingTableException(col) |
|
| 492 |
if type_is_array: type_ = sql_gen.ArrayType(type_) |
|
| 493 |
else: |
|
| 494 |
table = sql_gen.Table('columns', 'information_schema')
|
|
| 495 |
cols = [sql_gen.Col('data_type'), sql_gen.Col('udt_name'),
|
|
| 496 |
'column_default', sql_gen.Cast('boolean',
|
|
| 497 |
sql_gen.Col('is_nullable'))]
|
|
| 498 |
|
|
| 499 |
conds = [('table_name', col.table.name),
|
|
| 500 |
('column_name', strings.ustr(col.name))]
|
|
| 501 |
schema = col.table.schema |
|
| 502 |
if schema != None: conds.append(('table_schema', schema))
|
|
| 503 |
|
|
| 504 |
cur = select(self, table, cols, conds, order_by='table_schema', |
|
| 505 |
limit=1, cacheable=cacheable, log_level=4) |
|
| 506 |
try: type_, extra_type, default, nullable = row(cur) |
|
| 507 |
except StopIteration: raise sql_gen.NoUnderlyingTableException(col) |
|
| 508 |
if type_ == 'USER-DEFINED': type_ = extra_type |
|
| 509 |
elif type_ == 'ARRAY': |
|
| 510 |
type_ = sql_gen.ArrayType(strings.remove_prefix('_', extra_type,
|
|
| 511 |
require=True)) |
|
| 476 | 512 |
|
| 477 |
conds = [('table_name', col.table.name),
|
|
| 478 |
('column_name', strings.ustr(col.name))]
|
|
| 479 |
schema = col.table.schema |
|
| 480 |
if schema != None: conds.append(('table_schema', schema))
|
|
| 481 |
|
|
| 482 |
cur = select(self, table, cols, conds, order_by='table_schema', limit=1, |
|
| 483 |
cacheable=cacheable, log_level=4) # TODO: order by search_path order |
|
| 484 |
try: type_, extra_type, default, nullable = row(cur) |
|
| 485 |
except StopIteration: raise sql_gen.NoUnderlyingTableException(col) |
|
| 486 |
default = sql_gen.as_Code(default, self) |
|
| 487 |
if type_ == 'USER-DEFINED': type_ = extra_type |
|
| 488 |
elif type_ == 'ARRAY': |
|
| 489 |
type_ = sql_gen.ArrayType(strings.remove_prefix('_', extra_type,
|
|
| 490 |
require=True)) |
|
| 491 |
|
|
| 513 |
if default != None: default = sql_gen.as_Code(default, self) |
|
| 492 | 514 |
return sql_gen.TypedCol(col.name, type_, default, nullable) |
| 493 | 515 |
|
| 494 | 516 |
def TempFunction(self, name): |
Also available in: Unified diff
sql.py: DbConn.col_info(): For PostgreSQL, use pg_catalog tables directly instead of their views in information_schema. This allows using ::regclass to look up the table in the search_path, and fixes a bug in imports with an explicit public schema where column types were looked up in public instead of public.<version>. Also don't wrap default using sql_gen.as_Code() when it's None (indicating no default value, aka default=NULL), because this value is interpreted specially by sql_gen.TypedCol.