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-2025 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 

25import re 

26 

27from .base import MSDialect 

28from .base import MSIdentifierPreparer 

29from ... import types as sqltypes 

30from ... import util 

31from ...engine import processors 

32 

33 

34class _MSNumeric_pymssql(sqltypes.Numeric): 

35 def result_processor(self, dialect, type_): 

36 if not self.asdecimal: 

37 return processors.to_float 

38 else: 

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

40 

41 

42class MSIdentifierPreparer_pymssql(MSIdentifierPreparer): 

43 def __init__(self, dialect): 

44 super().__init__(dialect) 

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

46 # yet does not require that percent signs be doubled 

47 self._double_percents = False 

48 

49 

50class MSDialect_pymssql(MSDialect): 

51 supports_statement_cache = True 

52 supports_native_decimal = True 

53 supports_native_uuid = True 

54 driver = "pymssql" 

55 

56 preparer = MSIdentifierPreparer_pymssql 

57 

58 colspecs = util.update_copy( 

59 MSDialect.colspecs, 

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

61 ) 

62 

63 @classmethod 

64 def import_dbapi(cls): 

65 module = __import__("pymssql") 

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

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

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

69 # TODO: monkeypatching here is less than ideal 

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

71 

72 if client_ver < (1,): 

73 util.warn( 

74 "The pymssql dialect expects at least " 

75 "the 1.0 series of the pymssql DBAPI." 

76 ) 

77 return module 

78 

79 def _get_server_version_info(self, connection): 

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

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

82 if m: 

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

84 else: 

85 return None 

86 

87 def create_connect_args(self, url): 

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

89 opts.update(url.query) 

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

91 if port and "host" in opts: 

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

93 return ([], opts) 

94 

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

96 for msg in ( 

97 "Adaptive Server connection timed out", 

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

99 "message 20003", # connection timeout 

100 "Error 10054", 

101 "Not connected to any MS SQL server", 

102 "Connection is closed", 

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

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

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

106 "The server failed to resume the transaction", 

107 ): 

108 if msg in str(e): 

109 return True 

110 else: 

111 return False 

112 

113 def get_isolation_level_values(self, dbapi_connection): 

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

115 "AUTOCOMMIT" 

116 ] 

117 

118 def set_isolation_level(self, dbapi_connection, level): 

119 if level == "AUTOCOMMIT": 

120 dbapi_connection.autocommit(True) 

121 else: 

122 dbapi_connection.autocommit(False) 

123 super().set_isolation_level(dbapi_connection, level) 

124 

125 

126dialect = MSDialect_pymssql