Python源码示例：sqlalchemy.dialects.postgresql.INTEGER
示例1
def autogen_column_reflect(self, inspector, table, column_info):
if column_info.get('default') and \
isinstance(column_info['type'], (INTEGER, BIGINT)):
seq_match = re.match(
r"nextval\('(.+?)'::regclass\)",
column_info['default'])
if seq_match:
info = inspector.bind.execute(text(
"select c.relname, a.attname "
"from pg_class as c join pg_depend d on d.objid=c.oid and "
"d.classid='pg_class'::regclass and "
"d.refclassid='pg_class'::regclass "
"join pg_class t on t.oid=d.refobjid "
"join pg_attribute a on a.attrelid=t.oid and "
"a.attnum=d.refobjsubid "
"where c.relkind='S' and c.relname=:seqname"
), seqname=seq_match.group(1)).first()
if info:
seqname, colname = info
if colname == column_info['name']:
log.info(
"Detected sequence named '%s' as "
"owned by integer column '%s(%s)', "
"assuming SERIAL and omitting",
seqname, table.name, colname)
# sequence, and the owner is this column,
# its a SERIAL - whack it!
del column_info['default']
示例2
def autogen_column_reflect(self, inspector, table, column_info):
if column_info.get("default") and isinstance(
column_info["type"], (INTEGER, BIGINT)
):
seq_match = re.match(
r"nextval\('(.+?)'::regclass\)", column_info["default"]
)
if seq_match:
info = sqla_compat._exec_on_inspector(
inspector,
text(
"select c.relname, a.attname "
"from pg_class as c join "
"pg_depend d on d.objid=c.oid and "
"d.classid='pg_class'::regclass and "
"d.refclassid='pg_class'::regclass "
"join pg_class t on t.oid=d.refobjid "
"join pg_attribute a on a.attrelid=t.oid and "
"a.attnum=d.refobjsubid "
"where c.relkind='S' and c.relname=:seqname"
),
seqname=seq_match.group(1),
).first()
if info:
seqname, colname = info
if colname == column_info["name"]:
log.info(
"Detected sequence named '%s' as "
"owned by integer column '%s(%s)', "
"assuming SERIAL and omitting",
seqname,
table.name,
colname,
)
# sequence, and the owner is this column,
# its a SERIAL - whack it!
del column_info["default"]
示例3
def setup_class(cls):
con = testing.db.connect()
for ddl in [
'CREATE SCHEMA "SomeSchema"',
"CREATE DOMAIN testdomain INTEGER NOT NULL DEFAULT 42",
"CREATE DOMAIN test_schema.testdomain INTEGER DEFAULT 0",
"CREATE TYPE testtype AS ENUM ('test')",
"CREATE DOMAIN enumdomain AS testtype",
"CREATE DOMAIN arraydomain AS INTEGER[]",
'CREATE DOMAIN "SomeSchema"."Quoted.Domain" INTEGER DEFAULT 0',
]:
try:
con.exec_driver_sql(ddl)
except exc.DBAPIError as e:
if "already exists" not in str(e):
raise e
con.exec_driver_sql(
"CREATE TABLE testtable (question integer, answer " "testdomain)"
)
con.exec_driver_sql(
"CREATE TABLE test_schema.testtable(question "
"integer, answer test_schema.testdomain, anything "
"integer)"
)
con.exec_driver_sql(
"CREATE TABLE crosschema (question integer, answer "
"test_schema.testdomain)"
)
con.exec_driver_sql(
"CREATE TABLE enum_test (id integer, data enumdomain)"
)
con.exec_driver_sql(
"CREATE TABLE array_test (id integer, data arraydomain)"
)
con.exec_driver_sql(
"CREATE TABLE quote_test "
'(id integer, data "SomeSchema"."Quoted.Domain")'
)
示例4
def test_array_domain_is_reflected(self):
metadata = MetaData(testing.db)
table = Table("array_test", metadata, autoload=True)
eq_(table.c.data.type.__class__, ARRAY)
eq_(table.c.data.type.item_type.__class__, INTEGER)
示例5
def test_quoted_remote_schema_domain_is_reflected(self):
metadata = MetaData(testing.db)
table = Table("quote_test", metadata, autoload=True)
eq_(table.c.data.type.__class__, INTEGER)
示例6
def autogen_column_reflect(self, inspector, table, column_info):
if column_info.get('default') and \
isinstance(column_info['type'], (INTEGER, BIGINT)):
seq_match = re.match(
r"nextval\('(.+?)'::regclass\)",
column_info['default'])
if seq_match:
info = inspector.bind.execute(text(
"select c.relname, a.attname "
"from pg_class as c join pg_depend d on d.objid=c.oid and "
"d.classid='pg_class'::regclass and "
"d.refclassid='pg_class'::regclass "
"join pg_class t on t.oid=d.refobjid "
"join pg_attribute a on a.attrelid=t.oid and "
"a.attnum=d.refobjsubid "
"where c.relkind='S' and c.relname=:seqname"
), seqname=seq_match.group(1)).first()
if info:
seqname, colname = info
if colname == column_info['name']:
log.info(
"Detected sequence named '%s' as "
"owned by integer column '%s(%s)', "
"assuming SERIAL and omitting",
seqname, table.name, colname)
# sequence, and the owner is this column,
# its a SERIAL - whack it!
del column_info['default']
