Source code for pydal.adapters.informix

from ..helpers.classes import ConnectionConfigurationMixin
from .base import SQLAdapter
from . import adapters, with_connection_or_raise


[docs]@adapters.register_for('informix') class Informix(ConnectionConfigurationMixin, SQLAdapter): dbengine = 'informix' drivers = ('informixdb',) def _initialize_(self, do_connect): super(Informix, self)._initialize_(do_connect) ruri = self.uri.split('://', 1)[1] m = self.REGEX_URI.match(ruri) if not m: raise SyntaxError("Invalid URI string in DAL") user = self.credential_decoder(m.group('user')) if not user: raise SyntaxError('User required') password = self.credential_decoder(m.group('password')) if not password: password = '' host = m.group('host') if not host: raise SyntaxError('Host name required') db = m.group('db') if not db: raise SyntaxError('Database name required') self.dsn = '%s@%s' % (db, host) self.driver_args.update(user=user, password=password) self._mock_reconnect()
[docs] def connector(self): return self.driver.connect(self.dsn, **self.driver_args)
def _configure_on_first_reconnect(self): self.dbms_version = int(self.connection.dbms_version.split('.')[0]) @with_connection_or_raise def execute(self, *args, **kwargs): command = self.filter_sql_command(args[0]) if command[-1:] == ';': command = command[:-1] handlers = self._build_handlers_for_execution() for handler in handlers: handler.before_execute(command) rv = self.cursor.execute(command, *args[1:], **kwargs) for handler in handlers: handler.after_execute(command) return rv
[docs] def test_connection(self): self.execute('SELECT COUNT(*) FROM systables;')
[docs] def lastrowid(self, table): return self.cursor.sqlerrd[1]
[docs]@adapters.register_for('informix-se') class InformixSE(Informix):
[docs] def rowslice(self, rows, minimum=0, maximum=None): if maximum is None: return rows[minimum:] return rows[minimum:maximum]