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 @classmethod 

65 def import_dbapi(cls): 

66 module = __import__("pymssql") 

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

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

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

70 # TODO: monkeypatching here is less than ideal 

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

72 

73 if client_ver < (1,): 

74 util.warn( 

75 "The pymssql dialect expects at least " 

76 "the 1.0 series of the pymssql DBAPI." 

77 ) 

78 return module 

79 

80 def _get_server_version_info(self, connection): 

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

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

83 if m: 

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

85 else: 

86 return None 

87 

88 def create_connect_args(self, url): 

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

90 opts.update(url.query) 

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

92 if port and "host" in opts: 

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

94 return ([], opts) 

95 

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

97 for msg in ( 

98 "Adaptive Server connection timed out", 

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

100 "message 20003", # connection timeout 

101 "Error 10054", 

102 "Not connected to any MS SQL server", 

103 "Connection is closed", 

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

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

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

107 "The server failed to resume the transaction", 

108 ): 

109 if msg in str(e): 

110 return True 

111 else: 

112 return False 

113 

114 def get_isolation_level_values(self, dbapi_connection): 

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

116 "AUTOCOMMIT" 

117 ] 

118 

119 def set_isolation_level(self, dbapi_connection, level): 

120 if level == "AUTOCOMMIT": 

121 dbapi_connection.autocommit(True) 

122 else: 

123 dbapi_connection.autocommit(False) 

124 super().set_isolation_level(dbapi_connection, level) 

125 

126 

127dialect = MSDialect_pymssql