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

1# mssql/pymssql.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 

8""" 

9.. dialect:: mssql+pymssql 

10 :name: pymssql 

11 :dbapi: pymssql 

12 :connectstring: mssql+pymssql://<username>:<password>@<freetds_name>/?charset=utf8 

13 

14pymssql is a Python module that provides a Python DBAPI interface around 

15`FreeTDS <https://www.freetds.org/>`_. 

16 

17.. note:: 

18 

19 pymssql is currently not included in SQLAlchemy's continuous integration 

20 (CI) testing. 

21 

22 

23""" # noqa 

24import re 

25 

26from .base import MSDialect 

27from .base import MSIdentifierPreparer 

28from ... import processors 

29from ... import types as sqltypes 

30from ... import util 

31 

32 

33class _MSNumeric_pymssql(sqltypes.Numeric): 

34 def result_processor(self, dialect, type_): 

35 if not self.asdecimal: 

36 return processors.to_float 

37 else: 

38 return sqltypes.Numeric.result_processor(self, dialect, type_) 

39 

40 

41class MSIdentifierPreparer_pymssql(MSIdentifierPreparer): 

42 def __init__(self, dialect): 

43 super(MSIdentifierPreparer_pymssql, self).__init__(dialect) 

44 # pymssql has the very unusual behavior that it uses pyformat 

45 # yet does not require that percent signs be doubled 

46 self._double_percents = False 

47 

48 

49class MSDialect_pymssql(MSDialect): 

50 supports_statement_cache = True 

51 supports_native_decimal = True 

52 driver = "pymssql" 

53 

54 preparer = MSIdentifierPreparer_pymssql 

55 

56 colspecs = util.update_copy( 

57 MSDialect.colspecs, 

58 {sqltypes.Numeric: _MSNumeric_pymssql, sqltypes.Float: sqltypes.Float}, 

59 ) 

60 

61 @classmethod 

62 def dbapi(cls): 

63 module = __import__("pymssql") 

64 # pymmsql < 2.1.1 doesn't have a Binary method. we use string 

65 client_ver = tuple(int(x) for x in module.__version__.split(".")) 

66 if client_ver < (2, 1, 1): 

67 # TODO: monkeypatching here is less than ideal 

68 module.Binary = lambda x: x if hasattr(x, "decode") else str(x) 

69 

70 if client_ver < (1,): 

71 util.warn( 

72 "The pymssql dialect expects at least " 

73 "the 1.0 series of the pymssql DBAPI." 

74 ) 

75 return module 

76 

77 def _get_server_version_info(self, connection): 

78 vers = connection.exec_driver_sql("select @@version").scalar() 

79 m = re.match(r"Microsoft .*? - (\d+)\.(\d+)\.(\d+)\.(\d+)", vers) 

80 if m: 

81 return tuple(int(x) for x in m.group(1, 2, 3, 4)) 

82 else: 

83 return None 

84 

85 def create_connect_args(self, url): 

86 opts = url.translate_connect_args(username="user") 

87 opts.update(url.query) 

88 port = opts.pop("port", None) 

89 if port and "host" in opts: 

90 opts["host"] = "%s:%s" % (opts["host"], port) 

91 return [[], opts] 

92 

93 def is_disconnect(self, e, connection, cursor): 

94 for msg in ( 

95 "Adaptive Server connection timed out", 

96 "Net-Lib error during Connection reset by peer", 

97 "message 20003", # connection timeout 

98 "Error 10054", 

99 "Not connected to any MS SQL server", 

100 "Connection is closed", 

101 "message 20006", # Write to the server failed 

102 "message 20017", # Unexpected EOF from the server 

103 "message 20047", # DBPROCESS is dead or not enabled 

104 ): 

105 if msg in str(e): 

106 return True 

107 else: 

108 return False 

109 

110 def set_isolation_level(self, connection, level): 

111 if level == "AUTOCOMMIT": 

112 connection.autocommit(True) 

113 else: 

114 connection.autocommit(False) 

115 super(MSDialect_pymssql, self).set_isolation_level( 

116 connection, level 

117 ) 

118 

119 

120dialect = MSDialect_pymssql