1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
# +--------------------------------------------------------------------------+
# | Licensed Materials - Property of IBM |
# | |
# | (C) Copyright IBM Corporation 2008, 2013. |
# +--------------------------------------------------------------------------+
# | This module complies with SQLAlchemy 0.8 and is |
# | Licensed under the Apache License, Version 2.0 (the "License"); |
# | you may not use this file except in compliance with the License. |
# | You may obtain a copy of the License at |
# | http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable |
# | law or agreed to in writing, software distributed under the License is |
# | distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
# | KIND, either express or implied. See the License for the specific |
# | language governing permissions and limitations under the License. |
# +--------------------------------------------------------------------------+
# | Authors: Jaimy Azle, Rahul Priyadarshi |
# | Contributors: Mike Bayer |
# +--------------------------------------------------------------------------+
from sqlalchemy import util
import urllib
from sqlalchemy.connectors.pyodbc import PyODBCConnector
from .base import _SelectLastRowIDMixin, DB2ExecutionContext, DB2Dialect
class DB2ExecutionContext_pyodbc(_SelectLastRowIDMixin, DB2ExecutionContext):
pass
class DB2Dialect_pyodbc(PyODBCConnector, DB2Dialect):
supports_unicode_statements = False
supports_native_decimal = True
supports_char_length = True
supports_native_decimal = False
execution_ctx_cls = DB2ExecutionContext_pyodbc
pyodbc_driver_name = "IBM DB2 ODBC DRIVER"
def create_connect_args(self, url):
opts = url.translate_connect_args(username='user')
opts.update(url.query)
keys = opts
query = url.query
connect_args = {}
for param in ('ansi', 'unicode_results', 'autocommit'):
if param in keys:
connect_args[param] = util.asbool(keys.pop(param))
if 'odbc_connect' in keys:
connectors = [urllib.unquote_plus(keys.pop('odbc_connect'))]
else:
dsn_connection = 'dsn' in keys or \
('host' in keys and 'database' not in keys)
if dsn_connection:
connectors = ['dsn=%s' % (keys.pop('host', '') or \
keys.pop('dsn', ''))]
else:
port = ''
if 'port' in keys and not 'port' in query:
port = '%d' % int(keys.pop('port'))
database = keys.pop('database', '')
connectors = ["DRIVER={%s}" %
keys.pop('driver', self.pyodbc_driver_name),
'hostname=%s;port=%s' % (keys.pop('host', ''), port),
'database=%s' % database]
user = keys.pop("user", None)
if user:
connectors.append("uid=%s" % user)
connectors.append("pwd=%s" % keys.pop('password', ''))
else:
connectors.append("trusted_connection=yes")
# if set to 'yes', the odbc layer will try to automagically
# convert textual data from your database encoding to your
# client encoding. this should obviously be set to 'no' if
# you query a cp1253 encoded database from a latin1 client...
if 'odbc_autotranslate' in keys:
connectors.append("autotranslate=%s" %
keys.pop("odbc_autotranslate"))
connectors.extend(['%s=%s' % (k, v)
for k, v in keys.iteritems()])
return [[";".join(connectors)], connect_args]
class AS400Dialect_pyodbc(PyODBCConnector, DB2Dialect):
supports_unicode_statements = False
supports_sane_rowcount = False
supports_sane_multi_rowcount = False
supports_native_decimal = True
supports_char_length = True
supports_native_decimal = False
pyodbc_driver_name = "IBM DB2 ODBC DRIVER"
|
