File: MySQLdb.py

package info (click to toggle)
python-mysqldb 0.1.1-1
  • links: PTS
  • area: main
  • in suites: potato
  • size: 264 kB
  • ctags: 244
  • sloc: ansic: 1,320; python: 1,079; makefile: 47; sh: 14
file content (272 lines) | stat: -rw-r--r-- 9,046 bytes parent folder | download
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
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
"""MySQLdb - A DB API v2.0 compatible interface to MySQL.

This module is a thin wrapper around _mysql, which mostly implements the
MySQL C API. All symbols from that module are imported.

connect() -- connects to server
type_conv -- dictionary mapping SQL types to Python functions, which
             convert a string into an appropriate data type. Reasonable
             defaults are set for most items, and you can add your own.

See the API specification and the MySQL documentation for more info
on other items.

This module uses the mxDateTime package for handling date/time types.
"""

__version__ = """$Revision: 1.12 $"""[11:-2]

from _mysql import *
from time import localtime
import re

threadsafety = 1
apilevel = "2.0"
paramstyle = "format"

type_conv = { FIELD_TYPE.TINY: int,
              FIELD_TYPE.SHORT: int,
              FIELD_TYPE.LONG: int,
              FIELD_TYPE.FLOAT: float,
              FIELD_TYPE.DOUBLE: float,
              FIELD_TYPE.LONGLONG: long,
              FIELD_TYPE.INT24: int,
              FIELD_TYPE.YEAR: int }

try:
    from DateTime import Date, Time, Timestamp, ISO

    def DateFromTicks(ticks):
	return apply(Date, localtime(ticks)[:3])

    def TimeFromTicks(ticks):
	return apply(Time, localtime(ticks)[3:6])

    def TimestampFromTicks(ticks):
	return apply(Timestamp, localtime(ticks)[:6])

    def format_DATE(d):      return d.Format("%Y-%m-%d")
    def format_TIME(d):      return d.Format("%H:%M:%S")
    def format_TIMESTAMP(d): return d.Format("%Y-%m-%d %H:%M:%S")

    def mysql_timestamp_converter(s):
	parts = map(int, filter(None, (s[:4],s[4:6],s[6:8],
				       s[8:10],s[10:12],s[12:14])))
	return apply(Timestamp, tuple(parts))

    type_conv[FIELD_TYPE.TIMESTAMP] = mysql_timestamp_converter
    type_conv[FIELD_TYPE.DATETIME] = ISO.ParseDateTime
    #type_conv[FIELD_TYPE.TIME] = ISO.ParseTime
    type_conv[FIELD_TYPE.DATE] = ISO.ParseDate

except ImportError:
    # no DateTime? We'll muddle through somehow.
    from time import strftime

    def DateFromTicks(ticks):
	return strftime("%Y-%m-%d", localtime(ticks))

    def TimeFromTicks(ticks):
	return strftime("%H:%M:%S", localtime(ticks))

    def TimestampFromTicks(ticks):
	return strftime("%Y-%m-%d %H:%M:%S", localtime(ticks))

    def format_DATE(d): return d
    format_TIME = format_TIMESTAMP = format_DATE


class DBAPITypeObject:
    
    def __init__(self,*values):
        self.values = values
        
    def __cmp__(self,other):
        if other in self.values:
            return 0
        if other < self.values:
            return 1
        else:
            return -1


_Set = DBAPITypeObject

STRING    = _Set(FIELD_TYPE.CHAR, FIELD_TYPE.ENUM, FIELD_TYPE.INTERVAL,
                 FIELD_TYPE.SET, FIELD_TYPE.STRING, FIELD_TYPE.VAR_STRING)
BINARY    = _Set(FIELD_TYPE.BLOB, FIELD_TYPE.LONG_BLOB, FIELD_TYPE.MEDIUM_BLOB,
                 FIELD_TYPE.TINY_BLOB)
NUMBER    = _Set(FIELD_TYPE.DECIMAL, FIELD_TYPE.DOUBLE, FIELD_TYPE.FLOAT,
	         FIELD_TYPE.INT24, FIELD_TYPE.LONG, FIELD_TYPE.LONGLONG,
	         FIELD_TYPE.TINY, FIELD_TYPE.YEAR)
DATE      = _Set(FIELD_TYPE.DATE, FIELD_TYPE.NEWDATE)
TIME      = _Set(FIELD_TYPE.TIME)
TIMESTAMP = _Set(FIELD_TYPE.TIMESTAMP, FIELD_TYPE.DATETIME)
ROWID     = _Set()

def Binary(x): return str(x)

insert_values = re.compile(r'values\s(\(.+\))', re.IGNORECASE)

def escape_dict(d):
    d2 = {}
    for k,v in d.items():
        if v is None: d2[k] = "NULL"
        else: d2[k] = "'%s'" % escape_string(str(v))
    return d2


class _Cursor:
    
    """Created by a Connection object. Useful attributes:
    
    description -- DB API 7-tuple describing columns in last query
    arraysize -- default number of rows fetchmany() will fetch
    warnings -- should MySQL warnings raise a Warning exception?
    use -- should mysql_use_result be used instead of mysql_store_result?
    
    By default, warnings are issued, and mysql_store_result is used.
    See the MySQL docs for more information."""

    def __init__(self, connection, name='', use=0, warnings=1):
        self.connection = connection
        self.name = name
        self.description = None
        self.rowcount = -1
        self.result = None
        self.arraysize = 100
        self.warnings = warnings
	self.use = use
        
    def setinputsizes(self, *args): pass
      
    def setoutputsizes(self, *args): pass
         
    def execute(self, query, args=None):
        """cursor.execute(query, args=None)
        
        query -- string, query to execute on server
        args -- sequence or mapping, parameters to use with query."""
        from types import ListType, TupleType
        from string import rfind, join, split, atoi
        if not args:
            self._query(query)
        elif type(args) is ListType and type(args[0]) is TupleType:
 	    self.executemany(query, args) # deprecated
 	else:
            try:
                self._query(query % escape_row(args))
            except TypeError:
                self._query(query % escape_dict(args))

    def executemany(self, query, args):
        """cursor.executemany(self, query, args)
        
        query -- string, query to execute on server
        args -- sequence of sequences or mappings, parameters to use with
            query. The query must contain the clause "values ( ... )".
            The parenthetical portion will be repeated once for each
            item in the sequence.
        
        This method performs multiple-row inserts and similar queries."""
        from string import join
        m = insert_values.search(query)
        if not m: raise ProgrammingError, "can't find values"
        p = m.start(1)
        escape = escape_row
        try:
	    q = [query % escape(args[0])]
	except TypeError:
	    escape = escape_dict
	    q = [query % escape(args[0])]
        qv = query[p:]
        for a in args[1:]: q.append(qv % escape(a))
        self._query(join(q, ',\n'))

    def _query(self, q):
        from string import split, atoi
        db = self.connection.db
        db.query(q)
        if self.use: self.result = db.use_result()
        else:        self.result = db.store_result()
	if self.result:
     	    self.description = self.result.describe()
     	    self.rowcount = db.affected_rows()
     	else:
     	    self.description = None
     	    self.rowcount = -1
     	if self.warnings:
     	    w = db.info()
     	    if w:
     	        warnings = atoi(split(w)[-1])
     	        if warnings:
     	            raise Warning, w

    def fetchone(self):
        """Fetches a single row from the cursor."""
        try:
            return self.result.fetch_row()
        except AttributeError:
            raise ProgrammingError, "no query executed yet"
             
    def fetchmany(self, size=None):
        """cursor.fetchmany(size=cursor.arraysize)
        
        size -- integer, maximum number of rows to fetch."""
        return self.result.fetch_rows(size or self.arraysize)
         
    def fetchall(self):
        """Fetchs all available rows from the cursor."""
        return self.result.fetch_all_rows()
         
    def nextset(self): return None
     
     
class Connection:

    """Connection(host=NULL, user=NULL, passwd=NULL, db=NULL,
                  port=<MYSQL_PORT>, unix_socket=NULL, client_flag=0)
    
    Note: This interface uses keyword arguments exclusively.
    
    host -- string, host to connect to or NULL pointer (localhost)
    user -- string, user to connect as or NULL (your username)
    passwd -- string, password to use or NULL (no password)
    db -- string, database to use or NULL (no DB selected)
    port -- integer, TCP/IP port to connect to or default MySQL port
    unix_socket -- string, location of unix_socket to use or use TCP
    client_flags -- integer, flags to use or 0 (see MySQL docs)
    conv -- dictionary, maps MySQL FIELD_TYPE.* to Python functions which
            convert a string to the appropriate Python type
    
    Returns a Connection object.
    
    Useful attributes and methods:
    
    db -- connection object from _mysql. Good for accessing some of the
        MySQL-specific calls.
    close -- close the connection.
    cursor -- create a cursor (emulated) for executing queries.
    CursorClass -- class used to create cursors (_Cursor). If you subclass
        the Connection object, you will probably want to override this.
    """
    
    CursorClass = _Cursor
    
    def __init__(self, **kwargs):
        from _mysql import connect
        if not kwargs.has_key('conv'): kwargs['conv'] = type_conv.copy()
        self.db = apply(connect, (), kwargs)
     
    def close(self):
        """Close the connection. No further activity possible."""
        self.db.close()
         
    def commit(self): """Does nothing as there are no transactions."""
    
    def cursor(self, *args, **kwargs):
        """Create a cursor on which queries may be performed."""
        return apply(self.CursorClass, (self,)+args, kwargs)


Connect = connect = Connection