Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/sqlalchemy/dialects/mssql/information_schema.py: 80%

49 statements  

« prev     ^ index     » next       coverage.py v7.0.1, created at 2022-12-25 06:11 +0000

1# mssql/information_schema.py 

2# Copyright (C) 2005-2022 the SQLAlchemy authors and contributors 

3# <see AUTHORS file> 

4# 

5# This module is part of SQLAlchemy and is released under 

6# the MIT License: https://www.opensource.org/licenses/mit-license.php 

7 

8from ... import cast 

9from ... import Column 

10from ... import MetaData 

11from ... import Table 

12from ... import util 

13from ...ext.compiler import compiles 

14from ...sql import expression 

15from ...types import Boolean 

16from ...types import Integer 

17from ...types import Numeric 

18from ...types import String 

19from ...types import TypeDecorator 

20from ...types import Unicode 

21 

22 

23ischema = MetaData() 

24 

25 

26class CoerceUnicode(TypeDecorator): 

27 impl = Unicode 

28 cache_ok = True 

29 

30 def process_bind_param(self, value, dialect): 

31 if util.py2k and isinstance(value, util.binary_type): 

32 value = value.decode(dialect.encoding) 

33 return value 

34 

35 def bind_expression(self, bindvalue): 

36 return _cast_on_2005(bindvalue) 

37 

38 

39class _cast_on_2005(expression.ColumnElement): 

40 def __init__(self, bindvalue): 

41 self.bindvalue = bindvalue 

42 

43 

44@compiles(_cast_on_2005) 

45def _compile(element, compiler, **kw): 

46 from . import base 

47 

48 if ( 

49 compiler.dialect.server_version_info is None 

50 or compiler.dialect.server_version_info < base.MS_2005_VERSION 

51 ): 

52 return compiler.process(element.bindvalue, **kw) 

53 else: 

54 return compiler.process(cast(element.bindvalue, Unicode), **kw) 

55 

56 

57schemata = Table( 

58 "SCHEMATA", 

59 ischema, 

60 Column("CATALOG_NAME", CoerceUnicode, key="catalog_name"), 

61 Column("SCHEMA_NAME", CoerceUnicode, key="schema_name"), 

62 Column("SCHEMA_OWNER", CoerceUnicode, key="schema_owner"), 

63 schema="INFORMATION_SCHEMA", 

64) 

65 

66tables = Table( 

67 "TABLES", 

68 ischema, 

69 Column("TABLE_CATALOG", CoerceUnicode, key="table_catalog"), 

70 Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"), 

71 Column("TABLE_NAME", CoerceUnicode, key="table_name"), 

72 Column("TABLE_TYPE", CoerceUnicode, key="table_type"), 

73 schema="INFORMATION_SCHEMA", 

74) 

75 

76columns = Table( 

77 "COLUMNS", 

78 ischema, 

79 Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"), 

80 Column("TABLE_NAME", CoerceUnicode, key="table_name"), 

81 Column("COLUMN_NAME", CoerceUnicode, key="column_name"), 

82 Column("IS_NULLABLE", Integer, key="is_nullable"), 

83 Column("DATA_TYPE", String, key="data_type"), 

84 Column("ORDINAL_POSITION", Integer, key="ordinal_position"), 

85 Column( 

86 "CHARACTER_MAXIMUM_LENGTH", Integer, key="character_maximum_length" 

87 ), 

88 Column("NUMERIC_PRECISION", Integer, key="numeric_precision"), 

89 Column("NUMERIC_SCALE", Integer, key="numeric_scale"), 

90 Column("COLUMN_DEFAULT", Integer, key="column_default"), 

91 Column("COLLATION_NAME", String, key="collation_name"), 

92 schema="INFORMATION_SCHEMA", 

93) 

94 

95mssql_temp_table_columns = Table( 

96 "COLUMNS", 

97 ischema, 

98 Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"), 

99 Column("TABLE_NAME", CoerceUnicode, key="table_name"), 

100 Column("COLUMN_NAME", CoerceUnicode, key="column_name"), 

101 Column("IS_NULLABLE", Integer, key="is_nullable"), 

102 Column("DATA_TYPE", String, key="data_type"), 

103 Column("ORDINAL_POSITION", Integer, key="ordinal_position"), 

104 Column( 

105 "CHARACTER_MAXIMUM_LENGTH", Integer, key="character_maximum_length" 

106 ), 

107 Column("NUMERIC_PRECISION", Integer, key="numeric_precision"), 

108 Column("NUMERIC_SCALE", Integer, key="numeric_scale"), 

109 Column("COLUMN_DEFAULT", Integer, key="column_default"), 

110 Column("COLLATION_NAME", String, key="collation_name"), 

111 schema="tempdb.INFORMATION_SCHEMA", 

112) 

113 

114constraints = Table( 

115 "TABLE_CONSTRAINTS", 

116 ischema, 

117 Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"), 

118 Column("TABLE_NAME", CoerceUnicode, key="table_name"), 

119 Column("CONSTRAINT_NAME", CoerceUnicode, key="constraint_name"), 

120 Column("CONSTRAINT_TYPE", CoerceUnicode, key="constraint_type"), 

121 schema="INFORMATION_SCHEMA", 

122) 

123 

124column_constraints = Table( 

125 "CONSTRAINT_COLUMN_USAGE", 

126 ischema, 

127 Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"), 

128 Column("TABLE_NAME", CoerceUnicode, key="table_name"), 

129 Column("COLUMN_NAME", CoerceUnicode, key="column_name"), 

130 Column("CONSTRAINT_NAME", CoerceUnicode, key="constraint_name"), 

131 schema="INFORMATION_SCHEMA", 

132) 

133 

134key_constraints = Table( 

135 "KEY_COLUMN_USAGE", 

136 ischema, 

137 Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"), 

138 Column("TABLE_NAME", CoerceUnicode, key="table_name"), 

139 Column("COLUMN_NAME", CoerceUnicode, key="column_name"), 

140 Column("CONSTRAINT_NAME", CoerceUnicode, key="constraint_name"), 

141 Column("CONSTRAINT_SCHEMA", CoerceUnicode, key="constraint_schema"), 

142 Column("ORDINAL_POSITION", Integer, key="ordinal_position"), 

143 schema="INFORMATION_SCHEMA", 

144) 

145 

146ref_constraints = Table( 

147 "REFERENTIAL_CONSTRAINTS", 

148 ischema, 

149 Column("CONSTRAINT_CATALOG", CoerceUnicode, key="constraint_catalog"), 

150 Column("CONSTRAINT_SCHEMA", CoerceUnicode, key="constraint_schema"), 

151 Column("CONSTRAINT_NAME", CoerceUnicode, key="constraint_name"), 

152 # TODO: is CATLOG misspelled ? 

153 Column( 

154 "UNIQUE_CONSTRAINT_CATLOG", 

155 CoerceUnicode, 

156 key="unique_constraint_catalog", 

157 ), 

158 Column( 

159 "UNIQUE_CONSTRAINT_SCHEMA", 

160 CoerceUnicode, 

161 key="unique_constraint_schema", 

162 ), 

163 Column( 

164 "UNIQUE_CONSTRAINT_NAME", CoerceUnicode, key="unique_constraint_name" 

165 ), 

166 Column("MATCH_OPTION", String, key="match_option"), 

167 Column("UPDATE_RULE", String, key="update_rule"), 

168 Column("DELETE_RULE", String, key="delete_rule"), 

169 schema="INFORMATION_SCHEMA", 

170) 

171 

172views = Table( 

173 "VIEWS", 

174 ischema, 

175 Column("TABLE_CATALOG", CoerceUnicode, key="table_catalog"), 

176 Column("TABLE_SCHEMA", CoerceUnicode, key="table_schema"), 

177 Column("TABLE_NAME", CoerceUnicode, key="table_name"), 

178 Column("VIEW_DEFINITION", CoerceUnicode, key="view_definition"), 

179 Column("CHECK_OPTION", String, key="check_option"), 

180 Column("IS_UPDATABLE", String, key="is_updatable"), 

181 schema="INFORMATION_SCHEMA", 

182) 

183 

184computed_columns = Table( 

185 "computed_columns", 

186 ischema, 

187 Column("object_id", Integer), 

188 Column("name", CoerceUnicode), 

189 Column("is_computed", Boolean), 

190 Column("is_persisted", Boolean), 

191 Column("definition", CoerceUnicode), 

192 schema="sys", 

193) 

194 

195sequences = Table( 

196 "SEQUENCES", 

197 ischema, 

198 Column("SEQUENCE_CATALOG", CoerceUnicode, key="sequence_catalog"), 

199 Column("SEQUENCE_SCHEMA", CoerceUnicode, key="sequence_schema"), 

200 Column("SEQUENCE_NAME", CoerceUnicode, key="sequence_name"), 

201 schema="INFORMATION_SCHEMA", 

202) 

203 

204 

205class IdentitySqlVariant(TypeDecorator): 

206 r"""This type casts sql_variant columns in the identity_columns view 

207 to numeric. This is required because: 

208 

209 * pyodbc does not support sql_variant 

210 * pymssql under python 2 return the byte representation of the number, 

211 int 1 is returned as "\x01\x00\x00\x00". On python 3 it returns the 

212 correct value as string. 

213 """ 

214 impl = Unicode 

215 cache_ok = True 

216 

217 def column_expression(self, colexpr): 

218 return cast(colexpr, Numeric) 

219 

220 

221identity_columns = Table( 

222 "identity_columns", 

223 ischema, 

224 Column("object_id", Integer), 

225 Column("name", CoerceUnicode), 

226 Column("is_identity", Boolean), 

227 Column("seed_value", IdentitySqlVariant), 

228 Column("increment_value", IdentitySqlVariant), 

229 Column("last_value", IdentitySqlVariant), 

230 Column("is_not_for_replication", Boolean), 

231 schema="sys", 

232)