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

1# dialects/mssql/pymssql.py 

2# Copyright (C) 2005-2026 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# mypy: ignore-errors 

8 

9 

10""" 

11.. dialect:: mssql+pymssql 

12 :name: pymssql 

13 :dbapi: pymssql 

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

15 

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

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

18 

19.. versionchanged:: 2.0.5 

20 

21 pymssql was restored to SQLAlchemy's continuous integration testing 

22 

23 

24""" # noqa 

25 

26import re 

27 

28from .base import MSDialect 

29from .base import MSIdentifierPreparer 

30from ... import types as sqltypes 

31from ... import util 

32from ...engine import processors 

33 

34 

35class _MSNumeric_pymssql(sqltypes.Numeric): 

36 def result_processor(self, dialect, type_): 

37 if not self.asdecimal: 

38 return processors.to_float 

39 else: 

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

41 

42 

43class MSIdentifierPreparer_pymssql(MSIdentifierPreparer): 

44 def __init__(self, dialect): 

45 super().__init__(dialect) 

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

47 # yet does not require that percent signs be doubled 

48 self._double_percents = False 

49 

50 

51class MSDialect_pymssql(MSDialect): 

52 supports_statement_cache = True 

53 supports_native_decimal = True 

54 supports_native_uuid = True 

55 driver = "pymssql" 

56 

57 preparer = MSIdentifierPreparer_pymssql 

58 

59 colspecs = util.update_copy( 

60 MSDialect.colspecs, 

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

62 ) 

63 

64 def retrieve_dbapi_version(self, dbapi): 

65 return util.parse_version_string(getattr(dbapi, "__version__", None)) 

66 

67 @classmethod 

68 def import_dbapi(cls): 

69 module = __import__("pymssql") 

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

71 client_ver = util.parse_version_string(module.__version__) 

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

73 # TODO: monkeypatching here is less than ideal 

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

75 

76 if client_ver < (1,): 

77 util.warn( 

78 "The pymssql dialect expects at least " 

79 "the 1.0 series of the pymssql DBAPI." 

80 ) 

81 return module 

82 

83 def _get_server_version_info(self, connection): 

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

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

86 if m: 

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

88 else: 

89 return None 

90 

91 def create_connect_args(self, url): 

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

93 opts.update(url.query) 

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

95 if port and "host" in opts: 

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

97 return ([], opts) 

98 

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

100 for msg in ( 

101 "Adaptive Server connection timed out", 

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

103 "message 20003", # connection timeout 

104 "Error 10054", 

105 "Not connected to any MS SQL server", 

106 "Connection is closed", 

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

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

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

110 "The server failed to resume the transaction", 

111 ): 

112 if msg in str(e): 

113 return True 

114 else: 

115 return False 

116 

117 def get_isolation_level_values(self, dbapi_connection): 

118 return super().get_isolation_level_values(dbapi_connection) + [ 

119 "AUTOCOMMIT" 

120 ] 

121 

122 def set_isolation_level(self, dbapi_connection, level): 

123 if level == "AUTOCOMMIT": 

124 dbapi_connection.autocommit(True) 

125 else: 

126 dbapi_connection.autocommit(False) 

127 super().set_isolation_level(dbapi_connection, level) 

128 

129 

130dialect = MSDialect_pymssql