123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232 |
- # mssql/information_schema.py
- # Copyright (C) 2005-2022 the SQLAlchemy authors and contributors
- # <see AUTHORS file>
- #
- # This module is part of SQLAlchemy and is released under
- # the MIT License: https://www.opensource.org/licenses/mit-license.php
- from ... import cast
- from ... import Column
- from ... import MetaData
- from ... import Table
- from ... import util
- from ...ext.compiler import compiles
- from ...sql import expression
- from ...types import Boolean
- from ...types import Integer
- from ...types import Numeric
- from ...types import String
- from ...types import TypeDecorator
- from ...types import Unicode
- ischema = MetaData()
- class CoerceUnicode(TypeDecorator):
- impl = Unicode
- cache_ok = True
- def process_bind_param(self, value, dialect):
- if util.py2k and isinstance(value, util.binary_type):
- value = value.decode(dialect.encoding)
- return value
- def bind_expression(self, bindvalue):
- return _cast_on_2005(bindvalue)
- class _cast_on_2005(expression.ColumnElement):
- def __init__(self, bindvalue):
- self.bindvalue = bindvalue
- @compiles(_cast_on_2005)
- def _compile(element, compiler, **kw):
- from . import base
- if (
- compiler.dialect.server_version_info is None
- or compiler.dialect.server_version_info < base.MS_2005_VERSION
- ):
- return compiler.process(element.bindvalue, **kw)
- else:
- return compiler.process(cast(element.bindvalue, Unicode), **kw)
- schemata = Table(
- "SCHEMATA",
- ischema,
- Column("CATALOG_NAME", CoerceUnicode, key="catalog_name"),
- Column("SCHEMA_NAME", CoerceUnicode, key="schema_name"),
- Column("SCHEMA_OWNER", CoerceUnicode, key="schema_owner"),
- schema="INFORMATION_SCHEMA",
- )
- tables = Table(
- "TABLES",
- ischema,
- Column("TABLE_CATALOG", CoerceUnicode, key="table_catalog"),
- Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"),
- Column("TABLE_NAME", CoerceUnicode, key="table_name"),
- Column("TABLE_TYPE", CoerceUnicode, key="table_type"),
- schema="INFORMATION_SCHEMA",
- )
- columns = Table(
- "COLUMNS",
- ischema,
- Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"),
- Column("TABLE_NAME", CoerceUnicode, key="table_name"),
- Column("COLUMN_NAME", CoerceUnicode, key="column_name"),
- Column("IS_NULLABLE", Integer, key="is_nullable"),
- Column("DATA_TYPE", String, key="data_type"),
- Column("ORDINAL_POSITION", Integer, key="ordinal_position"),
- Column(
- "CHARACTER_MAXIMUM_LENGTH", Integer, key="character_maximum_length"
- ),
- Column("NUMERIC_PRECISION", Integer, key="numeric_precision"),
- Column("NUMERIC_SCALE", Integer, key="numeric_scale"),
- Column("COLUMN_DEFAULT", Integer, key="column_default"),
- Column("COLLATION_NAME", String, key="collation_name"),
- schema="INFORMATION_SCHEMA",
- )
- mssql_temp_table_columns = Table(
- "COLUMNS",
- ischema,
- Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"),
- Column("TABLE_NAME", CoerceUnicode, key="table_name"),
- Column("COLUMN_NAME", CoerceUnicode, key="column_name"),
- Column("IS_NULLABLE", Integer, key="is_nullable"),
- Column("DATA_TYPE", String, key="data_type"),
- Column("ORDINAL_POSITION", Integer, key="ordinal_position"),
- Column(
- "CHARACTER_MAXIMUM_LENGTH", Integer, key="character_maximum_length"
- ),
- Column("NUMERIC_PRECISION", Integer, key="numeric_precision"),
- Column("NUMERIC_SCALE", Integer, key="numeric_scale"),
- Column("COLUMN_DEFAULT", Integer, key="column_default"),
- Column("COLLATION_NAME", String, key="collation_name"),
- schema="tempdb.INFORMATION_SCHEMA",
- )
- constraints = Table(
- "TABLE_CONSTRAINTS",
- ischema,
- Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"),
- Column("TABLE_NAME", CoerceUnicode, key="table_name"),
- Column("CONSTRAINT_NAME", CoerceUnicode, key="constraint_name"),
- Column("CONSTRAINT_TYPE", CoerceUnicode, key="constraint_type"),
- schema="INFORMATION_SCHEMA",
- )
- column_constraints = Table(
- "CONSTRAINT_COLUMN_USAGE",
- ischema,
- Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"),
- Column("TABLE_NAME", CoerceUnicode, key="table_name"),
- Column("COLUMN_NAME", CoerceUnicode, key="column_name"),
- Column("CONSTRAINT_NAME", CoerceUnicode, key="constraint_name"),
- schema="INFORMATION_SCHEMA",
- )
- key_constraints = Table(
- "KEY_COLUMN_USAGE",
- ischema,
- Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"),
- Column("TABLE_NAME", CoerceUnicode, key="table_name"),
- Column("COLUMN_NAME", CoerceUnicode, key="column_name"),
- Column("CONSTRAINT_NAME", CoerceUnicode, key="constraint_name"),
- Column("CONSTRAINT_SCHEMA", CoerceUnicode, key="constraint_schema"),
- Column("ORDINAL_POSITION", Integer, key="ordinal_position"),
- schema="INFORMATION_SCHEMA",
- )
- ref_constraints = Table(
- "REFERENTIAL_CONSTRAINTS",
- ischema,
- Column("CONSTRAINT_CATALOG", CoerceUnicode, key="constraint_catalog"),
- Column("CONSTRAINT_SCHEMA", CoerceUnicode, key="constraint_schema"),
- Column("CONSTRAINT_NAME", CoerceUnicode, key="constraint_name"),
- # TODO: is CATLOG misspelled ?
- Column(
- "UNIQUE_CONSTRAINT_CATLOG",
- CoerceUnicode,
- key="unique_constraint_catalog",
- ),
- Column(
- "UNIQUE_CONSTRAINT_SCHEMA",
- CoerceUnicode,
- key="unique_constraint_schema",
- ),
- Column(
- "UNIQUE_CONSTRAINT_NAME", CoerceUnicode, key="unique_constraint_name"
- ),
- Column("MATCH_OPTION", String, key="match_option"),
- Column("UPDATE_RULE", String, key="update_rule"),
- Column("DELETE_RULE", String, key="delete_rule"),
- schema="INFORMATION_SCHEMA",
- )
- views = Table(
- "VIEWS",
- ischema,
- Column("TABLE_CATALOG", CoerceUnicode, key="table_catalog"),
- Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"),
- Column("TABLE_NAME", CoerceUnicode, key="table_name"),
- Column("VIEW_DEFINITION", CoerceUnicode, key="view_definition"),
- Column("CHECK_OPTION", String, key="check_option"),
- Column("IS_UPDATABLE", String, key="is_updatable"),
- schema="INFORMATION_SCHEMA",
- )
- computed_columns = Table(
- "computed_columns",
- ischema,
- Column("object_id", Integer),
- Column("name", CoerceUnicode),
- Column("is_computed", Boolean),
- Column("is_persisted", Boolean),
- Column("definition", CoerceUnicode),
- schema="sys",
- )
- sequences = Table(
- "SEQUENCES",
- ischema,
- Column("SEQUENCE_CATALOG", CoerceUnicode, key="sequence_catalog"),
- Column("SEQUENCE_SCHEMA", CoerceUnicode, key="sequence_schema"),
- Column("SEQUENCE_NAME", CoerceUnicode, key="sequence_name"),
- schema="INFORMATION_SCHEMA",
- )
- class IdentitySqlVariant(TypeDecorator):
- r"""This type casts sql_variant columns in the identity_columns view
- to numeric. This is required because:
- * pyodbc does not support sql_variant
- * pymssql under python 2 return the byte representation of the number,
- int 1 is returned as "\x01\x00\x00\x00". On python 3 it returns the
- correct value as string.
- """
- impl = Unicode
- cache_ok = True
- def column_expression(self, colexpr):
- return cast(colexpr, Numeric)
- identity_columns = Table(
- "identity_columns",
- ischema,
- Column("object_id", Integer),
- Column("name", CoerceUnicode),
- Column("is_identity", Boolean),
- Column("seed_value", IdentitySqlVariant),
- Column("increment_value", IdentitySqlVariant),
- Column("last_value", IdentitySqlVariant),
- Column("is_not_for_replication", Boolean),
- schema="sys",
- )
|