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

1# mssql/pyodbc.py

3# <see AUTHORS file>

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

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

7r"""

8.. dialect:: mssql+pyodbc

9 :name: PyODBC

10 :dbapi: pyodbc

11 :connectstring: mssql+pyodbc://<username>:<password>@<dsnname>

12 :url: https://pypi.org/project/pyodbc/

14Connecting to PyODBC

15--------------------

17The URL here is to be translated to PyODBC connection strings, as

18detailed in `ConnectionStrings <https://code.google.com/p/pyodbc/wiki/ConnectionStrings>`_.

20DSN Connections

21^^^^^^^^^^^^^^^

23A DSN connection in ODBC means that a pre-existing ODBC datasource is

24configured on the client machine. The application then specifies the name

25of this datasource, which encompasses details such as the specific ODBC driver

26in use as well as the network address of the database. Assuming a datasource

27is configured on the client, a basic DSN-based connection looks like::

29 engine = create_engine("mssql+pyodbc://scott:tiger@some_dsn")

31Which above, will pass the following connection string to PyODBC::

33 DSN=some_dsn;UID=scott;PWD=tiger

35If the username and password are omitted, the DSN form will also add

36the ``Trusted_Connection=yes`` directive to the ODBC string.

38Hostname Connections

39^^^^^^^^^^^^^^^^^^^^

41Hostname-based connections are also supported by pyodbc. These are often

42easier to use than a DSN and have the additional advantage that the specific

43database name to connect towards may be specified locally in the URL, rather

44than it being fixed as part of a datasource configuration.

46When using a hostname connection, the driver name must also be specified in the

47query parameters of the URL. As these names usually have spaces in them, the

48name must be URL encoded which means using plus signs for spaces::

50 engine = create_engine("mssql+pyodbc://scott:tiger@myhost:port/databasename?driver=ODBC+Driver+17+for+SQL+Server")

52The ``driver`` keyword is significant to the pyodbc dialect and must be

53specified in lowercase.

55Any other names passed in the query string are passed through in the pyodbc

56connect string, such as ``authentication``, ``TrustServerCertificate``, etc.

57Multiple keyword arguments must be separated by an ampersand (``&``); these

58will be translated to semicolons when the pyodbc connect string is generated

59internally::

61 e = create_engine(

62 "mssql+pyodbc://scott:tiger@mssql2017:1433/test?"

63 "driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes"

64 "&authentication=ActiveDirectoryIntegrated"

65 )

67The equivalent URL can be constructed using :class:`_sa.engine.URL`::

69 from sqlalchemy.engine import URL

70 connection_url = URL.create(

71 "mssql+pyodbc",

72 username="scott",

73 password="tiger",

74 host="mssql2017",

75 port=1433,

76 database="test",

77 query={

78 "driver": "ODBC Driver 18 for SQL Server",

79 "TrustServerCertificate": "yes",

80 "authentication": "ActiveDirectoryIntegrated",

81 },

82 )

85Pass through exact Pyodbc string

86^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

88A PyODBC connection string can also be sent in pyodbc's format directly, as

89specified in `the PyODBC documentation

90<https://github.com/mkleehammer/pyodbc/wiki/Connecting-to-databases>`_,

91using the parameter ``odbc_connect``. A :class:`_sa.engine.URL` object

92can help make this easier::

94 from sqlalchemy.engine import URL

95 connection_string = "DRIVER={SQL Server Native Client 10.0};SERVER=dagger;DATABASE=test;UID=user;PWD=password"

96 connection_url = URL.create("mssql+pyodbc", query={"odbc_connect": connection_string})

98 engine = create_engine(connection_url)

100.. _mssql_pyodbc_access_tokens:

101

102Connecting to databases with access tokens

103^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

104

105Some database servers are set up to only accept access tokens for login. For

106example, SQL Server allows the use of Azure Active Directory tokens to connect

107to databases. This requires creating a credential object using the

108``azure-identity`` library. More information about the authentication step can be

109found in `Microsoft's documentation

110<https://docs.microsoft.com/en-us/azure/developer/python/azure-sdk-authenticate?tabs=bash>`_.

111

112After getting an engine, the credentials need to be sent to ``pyodbc.connect``

113each time a connection is requested. One way to do this is to set up an event

114listener on the engine that adds the credential token to the dialect's connect

115call. This is discussed more generally in :ref:`engines_dynamic_tokens`. For

116SQL Server in particular, this is passed as an ODBC connection attribute with

117a data structure `described by Microsoft

118<https://docs.microsoft.com/en-us/sql/connect/odbc/using-azure-active-directory#authenticating-with-an-access-token>`_.

119

120The following code snippet will create an engine that connects to an Azure SQL

121database using Azure credentials::

122

123 import struct

124 from sqlalchemy import create_engine, event

125 from sqlalchemy.engine.url import URL

126 from azure import identity

127

128 SQL_COPT_SS_ACCESS_TOKEN = 1256 # Connection option for access tokens, as defined in msodbcsql.h

129 TOKEN_URL = "https://database.windows.net/" # The token URL for any Azure SQL database

130

131 connection_string = "mssql+pyodbc://@my-server.database.windows.net/myDb?driver=ODBC+Driver+17+for+SQL+Server"

132

133 engine = create_engine(connection_string)

134

135 azure_credentials = identity.DefaultAzureCredential()

136

137 @event.listens_for(engine, "do_connect")

138 def provide_token(dialect, conn_rec, cargs, cparams):

139 # remove the "Trusted_Connection" parameter that SQLAlchemy adds

140 cargs[0] = cargs[0].replace(";Trusted_Connection=Yes", "")

141

142 # create token credential

143 raw_token = azure_credentials.get_token(TOKEN_URL).token.encode("utf-16-le")

144 token_struct = struct.pack(f"<I{len(raw_token)}s", len(raw_token), raw_token)

145

146 # apply it to keyword arguments

147 cparams["attrs_before"] = {SQL_COPT_SS_ACCESS_TOKEN: token_struct}

148

149.. tip::

150

151 The ``Trusted_Connection`` token is currently added by the SQLAlchemy

152 pyodbc dialect when no username or password is present. This needs

153 to be removed per Microsoft's

154 `documentation for Azure access tokens

155 <https://docs.microsoft.com/en-us/sql/connect/odbc/using-azure-active-directory#authenticating-with-an-access-token>`_,

156 stating that a connection string when using an access token must not contain

157 ``UID``, ``PWD``, ``Authentication`` or ``Trusted_Connection`` parameters.

158

159.. _azure_synapse_ignore_no_transaction_on_rollback:

160

161Avoiding transaction-related exceptions on Azure Synapse Analytics

162^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

163

164Azure Synapse Analytics has a significant difference in its transaction

165handling compared to plain SQL Server; in some cases an error within a Synapse

166transaction can cause it to be arbitrarily terminated on the server side, which

167then causes the DBAPI ``.rollback()`` method (as well as ``.commit()``) to

168fail. The issue prevents the usual DBAPI contract of allowing ``.rollback()``

169to pass silently if no transaction is present as the driver does not expect

170this condition. The symptom of this failure is an exception with a message

171resembling 'No corresponding transaction found. (111214)' when attempting to

172emit a ``.rollback()`` after an operation had a failure of some kind.

173

174This specific case can be handled by passing ``ignore_no_transaction_on_rollback=True`` to

175the SQL Server dialect via the :func:`_sa.create_engine` function as follows::

176

177 engine = create_engine(connection_url, ignore_no_transaction_on_rollback=True)

178

179Using the above parameter, the dialect will catch ``ProgrammingError``

180exceptions raised during ``connection.rollback()`` and emit a warning

181if the error message contains code ``111214``, however will not raise

182an exception.

183

184.. versionadded:: 1.4.40 Added the

185 ``ignore_no_transaction_on_rollback=True`` parameter.

186

187Enable autocommit for Azure SQL Data Warehouse (DW) connections

188^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

189

190Azure SQL Data Warehouse does not support transactions,

191and that can cause problems with SQLAlchemy's "autobegin" (and implicit

192commit/rollback) behavior. We can avoid these problems by enabling autocommit

193at both the pyodbc and engine levels::

194

195 connection_url = sa.engine.URL.create(

196 "mssql+pyodbc",

197 username="scott",

198 password="tiger",

199 host="dw.azure.example.com",

200 database="mydb",

201 query={

202 "driver": "ODBC Driver 17 for SQL Server",

203 "autocommit": "True",

204 },

205 )

206

207 engine = create_engine(connection_url).execution_options(

208 isolation_level="AUTOCOMMIT"

209 )

210

211Avoiding sending large string parameters as TEXT/NTEXT

212^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

213

214By default, for historical reasons, Microsoft's ODBC drivers for SQL Server

215send long string parameters (greater than 4000 SBCS characters or 2000 Unicode

216characters) as TEXT/NTEXT values. TEXT and NTEXT have been deprecated for many

217years and are starting to cause compatibility issues with newer versions of

218SQL_Server/Azure. For example, see `this

219issue <https://github.com/mkleehammer/pyodbc/issues/835>`_.

220

221Starting with ODBC Driver 18 for SQL Server we can override the legacy

222behavior and pass long strings as varchar(max)/nvarchar(max) using the

223``LongAsMax=Yes`` connection string parameter::

224

225 connection_url = sa.engine.URL.create(

226 "mssql+pyodbc",

227 username="scott",

228 password="tiger",

229 host="mssqlserver.example.com",

230 database="mydb",

231 query={

232 "driver": "ODBC Driver 18 for SQL Server",

233 "LongAsMax": "Yes",

234 },

235 )

236

237

238Pyodbc Pooling / connection close behavior

239------------------------------------------

240

241PyODBC uses internal `pooling

242<https://github.com/mkleehammer/pyodbc/wiki/The-pyodbc-Module#pooling>`_ by

243default, which means connections will be longer lived than they are within

244SQLAlchemy itself. As SQLAlchemy has its own pooling behavior, it is often

245preferable to disable this behavior. This behavior can only be disabled

246globally at the PyODBC module level, **before** any connections are made::

247

248 import pyodbc

249

250 pyodbc.pooling = False

251

252 # don't use the engine before pooling is set to False

253 engine = create_engine("mssql+pyodbc://user:pass@dsn")

254

255If this variable is left at its default value of ``True``, **the application

256will continue to maintain active database connections**, even when the

257SQLAlchemy engine itself fully discards a connection or if the engine is

258disposed.

259

260.. seealso::

261

262 `pooling <https://github.com/mkleehammer/pyodbc/wiki/The-pyodbc-Module#pooling>`_ -

263 in the PyODBC documentation.

264

265Driver / Unicode Support

266-------------------------

267

268PyODBC works best with Microsoft ODBC drivers, particularly in the area

269of Unicode support on both Python 2 and Python 3.

270

271Using the FreeTDS ODBC drivers on Linux or OSX with PyODBC is **not**

272recommended; there have been historically many Unicode-related issues

273in this area, including before Microsoft offered ODBC drivers for Linux

274and OSX. Now that Microsoft offers drivers for all platforms, for

275PyODBC support these are recommended. FreeTDS remains relevant for

276non-ODBC drivers such as pymssql where it works very well.

277

278

279Rowcount Support

280----------------

281

282Pyodbc only has partial support for rowcount. See the notes at

283:ref:`mssql_rowcount_versioning` for important notes when using ORM

284versioning.

285

286.. _mssql_pyodbc_fastexecutemany:

287

288Fast Executemany Mode

289---------------------

290

291The Pyodbc driver has added support for a "fast executemany" mode of execution

292which greatly reduces round trips for a DBAPI ``executemany()`` call when using

293Microsoft ODBC drivers, for **limited size batches that fit in memory**. The

294feature is enabled by setting the flag ``.fast_executemany`` on the DBAPI

295cursor when an executemany call is to be used. The SQLAlchemy pyodbc SQL

296Server dialect supports setting this flag automatically when the

297``.fast_executemany`` flag is passed to

298:func:`_sa.create_engine` ; note that the ODBC driver must be the Microsoft

299driver in order to use this flag::

300

301 engine = create_engine(

302 "mssql+pyodbc://scott:tiger@mssql2017:1433/test?driver=ODBC+Driver+13+for+SQL+Server",

303 fast_executemany=True)

304

305.. warning:: The pyodbc fast_executemany mode **buffers all rows in memory** and is

306 not compatible with very large batches of data. A future version of SQLAlchemy

307 may support this flag as a per-execution option instead.

308

309.. versionadded:: 1.3

310

311.. seealso::

312

313 `fast executemany <https://github.com/mkleehammer/pyodbc/wiki/Features-beyond-the-DB-API#fast_executemany>`_

314 - on github

315

316.. _mssql_pyodbc_setinputsizes:

317

318Setinputsizes Support

319-----------------------

320

321The pyodbc ``cursor.setinputsizes()`` method can be used if necessary. To

322enable this hook, pass ``use_setinputsizes=True`` to :func:`_sa.create_engine`::

323

324 engine = create_engine("mssql+pyodbc://...", use_setinputsizes=True)

325

326The behavior of the hook can then be customized, as may be necessary

327particularly if fast_executemany is in use, via the

328:meth:`.DialectEvents.do_setinputsizes` hook. See that method for usage

329examples.

330

331.. versionchanged:: 1.4.1 The pyodbc dialects will not use setinputsizes

332 unless ``use_setinputsizes=True`` is passed.

333

334""" # noqa

335

336

337import datetime

338import decimal

339import re

340import struct

341

342from .base import BINARY

343from .base import DATETIMEOFFSET

344from .base import MSDialect

345from .base import MSExecutionContext

346from .base import VARBINARY

347from ... import exc

348from ... import types as sqltypes

349from ... import util

350from ...connectors.pyodbc import PyODBCConnector

351

352

353class _ms_numeric_pyodbc(object):

354

355 """Turns Decimals with adjusted() < 0 or > 7 into strings.

356

357 The routines here are needed for older pyodbc versions

358 as well as current mxODBC versions.

359

360 """

361

362 def bind_processor(self, dialect):

363

364 super_process = super(_ms_numeric_pyodbc, self).bind_processor(dialect)

365

366 if not dialect._need_decimal_fix:

367 return super_process

368

369 def process(value):

370 if self.asdecimal and isinstance(value, decimal.Decimal):

371 adjusted = value.adjusted()

372 if adjusted < 0:

373 return self._small_dec_to_string(value)

374 elif adjusted > 7:

375 return self._large_dec_to_string(value)

376

377 if super_process:

378 return super_process(value)

379 else:

380 return value

381

382 return process

383

384 # these routines needed for older versions of pyodbc.

385 # as of 2.1.8 this logic is integrated.

386

387 def _small_dec_to_string(self, value):

388 return "%s0.%s%s" % (

389 (value < 0 and "-" or ""),

390 "0" * (abs(value.adjusted()) - 1),

391 "".join([str(nint) for nint in value.as_tuple()[1]]),

392 )

393

394 def _large_dec_to_string(self, value):

395 _int = value.as_tuple()[1]

396 if "E" in str(value):

397 result = "%s%s%s" % (

398 (value < 0 and "-" or ""),

399 "".join([str(s) for s in _int]),

400 "0" * (value.adjusted() - (len(_int) - 1)),

401 )

402 else:

403 if (len(_int) - 1) > value.adjusted():

404 result = "%s%s.%s" % (

405 (value < 0 and "-" or ""),

406 "".join([str(s) for s in _int][0 : value.adjusted() + 1]),

407 "".join([str(s) for s in _int][value.adjusted() + 1 :]),

408 )

409 else:

410 result = "%s%s" % (

411 (value < 0 and "-" or ""),

412 "".join([str(s) for s in _int][0 : value.adjusted() + 1]),

413 )

414 return result

415

416

417class _MSNumeric_pyodbc(_ms_numeric_pyodbc, sqltypes.Numeric):

418 pass

419

420

421class _MSFloat_pyodbc(_ms_numeric_pyodbc, sqltypes.Float):

422 pass

423

424

425class _ms_binary_pyodbc(object):

426 """Wraps binary values in dialect-specific Binary wrapper.

427 If the value is null, return a pyodbc-specific BinaryNull

428 object to prevent pyODBC [and FreeTDS] from defaulting binary

429 NULL types to SQLWCHAR and causing implicit conversion errors.

430 """

431

432 def bind_processor(self, dialect):

433 if dialect.dbapi is None:

434 return None

435

436 DBAPIBinary = dialect.dbapi.Binary

437

438 def process(value):

439 if value is not None:

440 return DBAPIBinary(value)

441 else:

442 # pyodbc-specific

443 return dialect.dbapi.BinaryNull

444

445 return process

446

447

448class _ODBCDateTimeBindProcessor(object):

449 """Add bind processors to handle datetimeoffset behaviors"""

450

451 has_tz = False

452

453 def bind_processor(self, dialect):

454 def process(value):

455 if value is None:

456 return None

457 elif isinstance(value, util.string_types):

458 # if a string was passed directly, allow it through

459 return value

460 elif not value.tzinfo or (not self.timezone and not self.has_tz):

461 # for DateTime(timezone=False)

462 return value

463 else:

464 # for DATETIMEOFFSET or DateTime(timezone=True)

465 #

466 # Convert to string format required by T-SQL

467 dto_string = value.strftime("%Y-%m-%d %H:%M:%S.%f %z")

468 # offset needs a colon, e.g., -0700 -> -07:00

469 # "UTC offset in the form (+-)HHMM[SS[.ffffff]]"

470 # backend currently rejects seconds / fractional seconds

471 dto_string = re.sub(

472 r"([\+\-]\d{2})([\d\.]+)$", r"\1:\2", dto_string

473 )

474 return dto_string

475

476 return process

477

478

479class _ODBCDateTime(_ODBCDateTimeBindProcessor, sqltypes.DateTime):

480 pass

481

482

483class _ODBCDATETIMEOFFSET(_ODBCDateTimeBindProcessor, DATETIMEOFFSET):

484 has_tz = True

485

486

487class _VARBINARY_pyodbc(_ms_binary_pyodbc, VARBINARY):

488 pass

489

490

491class _BINARY_pyodbc(_ms_binary_pyodbc, BINARY):

492 pass

493

494

495class MSExecutionContext_pyodbc(MSExecutionContext):

496 _embedded_scope_identity = False

497

498 def pre_exec(self):

499 """where appropriate, issue "select scope_identity()" in the same

500 statement.

501

502 Background on why "scope_identity()" is preferable to "@@identity":

503 https://msdn.microsoft.com/en-us/library/ms190315.aspx

504

505 Background on why we attempt to embed "scope_identity()" into the same

506 statement as the INSERT:

507 https://code.google.com/p/pyodbc/wiki/FAQs#How_do_I_retrieve_autogenerated/identity_values?

508

509 """

510

511 super(MSExecutionContext_pyodbc, self).pre_exec()

512

513 # don't embed the scope_identity select into an

514 # "INSERT .. DEFAULT VALUES"

515 if (

516 self._select_lastrowid

517 and self.dialect.use_scope_identity

518 and len(self.parameters[0])

519 ):

520 self._embedded_scope_identity = True

521

522 self.statement += "; select scope_identity()"

523

524 def post_exec(self):

525 if self._embedded_scope_identity:

526 # Fetch the last inserted id from the manipulated statement

527 # We may have to skip over a number of result sets with

528 # no data (due to triggers, etc.)

529 while True:

530 try:

531 # fetchall() ensures the cursor is consumed

532 # without closing it (FreeTDS particularly)

533 row = self.cursor.fetchall()[0]

534 break

535 except self.dialect.dbapi.Error:

536 # no way around this - nextset() consumes the previous set

537 # so we need to just keep flipping

538 self.cursor.nextset()

539

540 self._lastrowid = int(row[0])

541 else:

542 super(MSExecutionContext_pyodbc, self).post_exec()

543

544

545class MSDialect_pyodbc(PyODBCConnector, MSDialect):

546 supports_statement_cache = True

547

548 # mssql still has problems with this on Linux

549 supports_sane_rowcount_returning = False

550

551 execution_ctx_cls = MSExecutionContext_pyodbc

552

553 colspecs = util.update_copy(

554 MSDialect.colspecs,

555 {

556 sqltypes.Numeric: _MSNumeric_pyodbc,

557 sqltypes.Float: _MSFloat_pyodbc,

558 BINARY: _BINARY_pyodbc,

559 # support DateTime(timezone=True)

560 sqltypes.DateTime: _ODBCDateTime,

561 DATETIMEOFFSET: _ODBCDATETIMEOFFSET,

562 # SQL Server dialect has a VARBINARY that is just to support

563 # "deprecate_large_types" w/ VARBINARY(max), but also we must

564 # handle the usual SQL standard VARBINARY

565 VARBINARY: _VARBINARY_pyodbc,

566 sqltypes.VARBINARY: _VARBINARY_pyodbc,

567 sqltypes.LargeBinary: _VARBINARY_pyodbc,

568 },

569 )

570

571 def __init__(

572 self, description_encoding=None, fast_executemany=False, **params

573 ):

574 if "description_encoding" in params:

575 self.description_encoding = params.pop("description_encoding")

576 super(MSDialect_pyodbc, self).__init__(**params)

577 self.use_scope_identity = (

578 self.use_scope_identity

579 and self.dbapi

580 and hasattr(self.dbapi.Cursor, "nextset")

581 )

582 self._need_decimal_fix = self.dbapi and self._dbapi_version() < (

583 2,

584 1,

585 8,

586 )

587 self.fast_executemany = fast_executemany

588

589 def _get_server_version_info(self, connection):

590 try:

591 # "Version of the instance of SQL Server, in the form

592 # of 'major.minor.build.revision'"

593 raw = connection.exec_driver_sql(

594 "SELECT CAST(SERVERPROPERTY('ProductVersion') AS VARCHAR)"

595 ).scalar()

596 except exc.DBAPIError:

597 # SQL Server docs indicate this function isn't present prior to

598 # 2008. Before we had the VARCHAR cast above, pyodbc would also

599 # fail on this query.

600 return super(MSDialect_pyodbc, self)._get_server_version_info(

601 connection, allow_chars=False

602 )

603 else:

604 version = []

605 r = re.compile(r"[.\-]")

606 for n in r.split(raw):

607 try:

608 version.append(int(n))

609 except ValueError:

610 pass

611 return tuple(version)

612

613 def on_connect(self):

614 super_ = super(MSDialect_pyodbc, self).on_connect()

615

616 def on_connect(conn):

617 if super_ is not None:

618 super_(conn)

619

620 self._setup_timestampoffset_type(conn)

621

622 return on_connect

623

624 def _setup_timestampoffset_type(self, connection):

625 # output converter function for datetimeoffset

626 def _handle_datetimeoffset(dto_value):

627 tup = struct.unpack("<6hI2h", dto_value)

628 return datetime.datetime(

629 tup[0],

630 tup[1],

631 tup[2],

632 tup[3],

633 tup[4],

634 tup[5],

635 tup[6] // 1000,

636 util.timezone(

637 datetime.timedelta(hours=tup[7], minutes=tup[8])

638 ),

639 )

640

641 odbc_SQL_SS_TIMESTAMPOFFSET = -155 # as defined in SQLNCLI.h

642 connection.add_output_converter(

643 odbc_SQL_SS_TIMESTAMPOFFSET, _handle_datetimeoffset

644 )

645

646 def do_executemany(self, cursor, statement, parameters, context=None):

647 if self.fast_executemany:

648 cursor.fast_executemany = True

649 super(MSDialect_pyodbc, self).do_executemany(

650 cursor, statement, parameters, context=context

651 )

652

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

654 if isinstance(e, self.dbapi.Error):

655 code = e.args[0]

656 if code in {

657 "08S01",

658 "01000",

659 "01002",

660 "08003",

661 "08007",

662 "08S02",

663 "08001",

664 "HYT00",

665 "HY010",

666 "10054",

667 }:

668 return True

669 return super(MSDialect_pyodbc, self).is_disconnect(

670 e, connection, cursor

671 )

672

673

674dialect = MSDialect_pyodbc