ll.orasql contains utilities for working with cx_Oracle:
It allows calling procedures with keyword arguments (via the
Procedureclass).Query results will be put into
Recordobjects, where database fields are accessible as object attributes.The
Connectionclass provides methods for iterating through the database metadata.Importing this module adds support for URLs with the scheme
oracletoll.url. Examples of these URLs are:oracle://user:pwd@db/ oracle://user:pwd@db/view/ oracle://user:pwd@db/view/USER_TABLES.sql oracle://sys:pwd:sysdba@db/
class SQLObjectNotFoundError(IOError):
def __init__(self, obj):
selfclass SQLNoSuchObjectError(Exception):
def __init__(self, name, owner):
selfdef __repr__(self):
selfdef __str__(self):
selfclass UnknownModeError(ValueError):
def __init__(self, mode):
selfdef __repr__(self):
selfdef __str__(self):
selfclass ConflictError(ValueError):
def __init__(self, object, message):
selfdef __repr__(self):
selfdef __str__(self):
selfclass Args(dict):
An Args object is a subclass of dict that is used for
passing arguments to procedures and functions. Both item and attribute access
(i.e. __getitem__ and __getattr__) are available. Names are
case insensitive.
def __init__(self, arg=None, **kwargs):
selfdef update(self, arg=None, **kwargs):
selfdef __getitem__(self, name):
selfdef __setitem__(self, name, value):
selfdef __delitem__(self, name):
selfdef __getattr__(self, name):
selfdef __setattr__(self, name, value):
selfdef __delattr__(self, name):
selfdef __repr__(self):
selfclass CLOBStream(object):
A CLOBStream object provides streamlike access to a CLOB.
def __init__(self, value, encoding):
selfdef readall(self):
selfRead all remaining data from the stream and return the decoded unicode object for the data.
def readchunk(self):
selfRead a chunk of data from the stream and return the decoded unicode object for the data. Reading is done in optimally sized chunks.
def read(self, size=None):
selfRead size bytes from to stream and return the decoded unicode object
for the data. If size is None, all remaining data will be
read.
def reset(self):
selfReset the stream, so that the next read call starts at the
beginning of the CLOB.
class BLOBStream(object):
A BLOBStream object provides streamlike access to a BLOB.
def __init__(self, value):
selfdef readall(self):
selfRead all remaining data from the stream and return it.
def readchunk(self):
selfRead a chunk of data from the stream and return it. Reading is done in optimally sized chunks.
def read(self, size=None):
selfRead size bytes from to stream and return them. If size is
None, all remaining data will be read.
def reset(self):
selfReset the stream, so that the next read call starts at the
beginning of the BLOB.
def _decodeclob(value, encoding, readlobs):
def _decodeblob(value, readlobs):
class RecordMaker(object):
def __init__(self, cursor):
selfdef __call__(self, *row):
selfdef STRING(self, value):
selfdef LONG_STRING(self, value):
selfdef FIXED_CHAR(self, value):
selfdef CLOB(self, value):
selfdef NCLOB(self, value):
selfdef BLOB(self, value):
selfdef DEFAULT(self, value):
selfclass Record(tuple):
A Record is a subclass of tuple that is used for storing
results of database fetches and procedure and function calls. Both item and
attribute access (i.e. __getitem__ and __getattr__) are
available. Field names are case insensitive.
def __getitem__(self, arg):
selfdef __getattr__(self, name):
selfdef get(self, name, default=None):
selfReturn the value for the field named name. If this field doesn't
exist in self, return default instead.
def __contains__(self, name):
selfdef iterkeys(self):
selfReturn an iterator over field names
def keys(self):
selfReturn a list of field names
def items(self):
selfReturn a list of (field name, field value) tuples.
def iteritems(self):
selfReturn an iterator over (field name, field value) tuples.
def __repr__(self):
selfclass SessionPool(OracleSessionPool):
SessionPool is a subclass of cx_Oracle.SessionPool.
def __init__(self, user, password, database, min, max, increment, connectiontype=None, threaded=False, getmode=1, homogeneous=True):
selfdef connectstring(self):
selfdef __repr__(self):
selfclass Connection(cx_Oracle.Connection):
Connection is a subclass of cx_Oracle.Connection.
def __init__(self, *args, **kwargs):
selfCreate a new connection. In addition to the parameters supported by
cx_Oracle.connect the following keyword argument is supported.
readlobs(bool or integer)If
readlobsisFalseall cursor fetches returnCLOBStreamorBLOBStreamobjects. Ifreadlobsis anintLOB`s with a maximum size of :var:`readlobswill be returned asstr/unicodeobjects. IfreadlobsisTrueallLOBvalues will be returned as strings.
def connectstring(self):
selfdef cursor(self, readlobs=None):
selfReturn a new cursor for this connection. For the meaning of
readlobs see __init__.
def __repr__(self):
selfdef itertables(self, owner=<object object at 0xf67a2c80>, mode='flat'):
selfGenerator that yields all table definitions in the current users schema
(or all users schemas). mode specifies the order in which tables
will be yielded:
"create"Create order, deleting records from the table in this order will not violate foreign key constraints.
"drop"Drop order, deleting records from the table in this order will not violate foreign key constraints.
"flat"Unordered.
schema specifies from which user tables should be yielded:
"user"Only tables belonging to the current user (and those objects these depend on) will be yielded (i.e. tables from the
USER_TABLESview)."all"All tables accessible to the current user will be yielded (i.e. tables from the
ALL_TABLESviews)."dba"All tables from all users will be yielded (i.e. tables from the
DBA_TABLESview, this requires the appropriate privileges).
Tables that are materialized views will be skipped in all cases.
def itersequences(self, owner=<object object at 0xf67a2c80>):
selfGenerator that yields sequences. owner can be None,
ALL (the default) or a user name.
def iterfks(self, owner=<object object at 0xf67a2c80>):
selfGenerator that yields all foreign key constraints. owner can be
None, ALL (the default) or a user name.
def iterprivileges(self, owner=<object object at 0xf67a2c80>):
selfGenerator that yields object privileges. owner can be None,
ALL (the default) or a user name.
def iterusers(self):
selfGenerator that yields all users.
def iterobjects(self, owner=<object object at 0xf67a2c80>, mode='create'):
selfGenerator that yields the sequences, tables, primary keys, foreign keys, comments, unique constraints, indexes, views, functions, procedures, packages and types in the current users schema (or all users schemas) in a specified order.
mode specifies the order in which objects will be yielded:
"create"Create order, i.e. recreating the objects in this order will not lead to errors.
"drop"Drop order, i.e. dropping the objects in this order will not lead to errors.
"flat"Unordered.
schema specifies from which schema objects should be yielded:
"user"Only objects belonging to the current user (and those objects these depend on) will be yielded.
"all"All objects accessible to the current user will be yielded.
"dba"All objects from all users will be yielded (this requires the appropriate privileges).
def _getobject(self, name, owner=None):
selfdef getobject(self, name, owner=None):
selfReturn the object named name from the schema. If owner is
None the current schema is queried, else the specified one is
used. name and owner are treated case insensitively.
def connect(*args, **kwargs):
Create a connection to the database and return a Connection object.
class Cursor(OracleCursor):
A subclass of the cursor class in cx_Oracle. The "execute" methods
support a unicode statement and unicode parameters (they will be encoded in
the client encoding before being passed to the database). The "fetch" methods
will return records as Record objects and string values and
CLOB values, if the cursors readlobs attribute has the
appropriate value) will be returned as unicode objects (except for
BLOB values). (Note that strings in the national character set
(and NCLOB values) are not supported).
def __init__(self, connection, readlobs=None):
selfReturn a new cursor for the connection connection. For the meaning
of readlobs see Connection.__init__.
def ddprefix(self):
selfReturn whether the user has access to the DBA_* views ("dba") or
not ("all").
def ddprefixargs(self):
selfReturn whether the user has access to the DBA_ARGUMENTS view
("dba") or not ("all").
def _encode(self, value):
selfdef execute(self, statement, parameters=None, **kwargs):
selfdef executemany(self, statement, parameters):
selfdef __repr__(self):
selfdef formatstring(value, latin1=False):
def makeurl(name):
class MixinNormalDates(object):
Mixin class that provides methods for determining creation and modification dates for objects.
def cdate(self, connection=None):
selfdef udate(self, connection=None):
selfclass MixinCodeDDL(object):
Mixin class that provides methods returning the create and drop statements for various objects.
def createddl(self, connection=None, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef fixname(cls, code):
clsdef getfullname(name, owner):
class Object(object):
The base class for all Python classes modelling schema objects in the database.
Subclasses are: Sequence, Table, PrimaryKey,
Comment, ForeignKey, Index, Unique,
Synonym, View, MaterializedView, Library,
Function, Package, Type, Trigger,
JavaSource and Column.
class __metaclass__(type):
def __init__(self, name, owner=None, connection=None):
selfdef __repr__(self):
selfdef __str__(self):
selfdef __eq__(self, other):
selfdef __ne__(self, other):
selfdef __hash__(self):
selfdef getfullname(self):
selfdef createddl(self, *args, **kwargs):
selfReturn SQL code to create this object.
def dropddl(self, *args, **kwargs):
selfReturn SQL code to drop this object
def fixname(self, *args, **kwargs):
selfReplace the name of the object in the SQL code code with
the name of self.
def cdate(self, *args, **kwargs):
selfReturn a datetime.datetime object with the creation date of
self in the database specified by connection (or
None if such information is not available).
def udate(self, *args, **kwargs):
selfReturn a datetime.datetime object with the last modification
date of self in the database specified by connection
(or None if such information is not available).
def iterreferences(self, connection=None):
selfObjects directly used by self.
If connection is not None it will be used as the database
connection from which to fetch data. If connection is None
the connection from which self has been extracted will be used. If
there is not such connection, you'll get an exception.
def iterreferencesall(self, connection=None, done=None):
selfAll objects used by self (recursively).
For the meaning of connection see iterreferences.
done is used internally and shouldn't be passed.
def iterreferencedby(self, connection=None):
selfObjects using self.
For the meaning of connection see iterreferences.
def iterreferencedbyall(self, connection=None, done=None):
selfAll objects depending on self (recursively).
For the meaning of connection see iterreferences.
done is used internally and shouldn't be passed.
def getconnection(self, connection):
selfdef getcursor(self, connection):
selfproperty connectstring:
def __get__(self):
def iternames(cls, connection, owner=<object object at 0xf67a2c80>):
clsGenerator that yields the names of all objects of this type. The argument
owner specifies whos objects are yielded:
NoneAll objects belonging to the current user (i.e. via the view
USER_OBJECTS).ALLAll objects for all users (via the views
ALL_OBJECTSorDBA_OBJECTS)- username (string)
All objects belonging the the specified user
def iterobjects(cls, connection, owner=<object object at 0xf67a2c80>):
clsGenerator that yields all objects of this type in the current users schema.
The argument owner specifies whos objects are yielded:
NoneAll objects belonging to the current user (i.e. via the view
USER_OBJECTS).ALLAll objects for all users (via the views
ALL_OBJECTSorDBA_OBJECTS)- username (string)
All objects belonging the the specified user
class Sequence(MixinNormalDates, Object):
Models a sequence in the database.
def _createddl(self, connection, term, copyvalue):
selfdef createddl(self, connection=None, term=True):
selfdef createddlcopy(self, connection=None, term=True):
selfReturn SQL code to create an identical copy of this sequence.
def dropddl(self, connection=None, term=True):
selfdef fixname(cls, code):
clsdef iterreferences(self, connection=None, done=None):
selfdef _columntype(rec, data_precision=None, data_scale=None, char_length=None):
def _columndefault(rec):
class Table(MixinNormalDates, Object):
Models a table in the database.
def createddl(self, connection=None, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef fixname(cls, code):
clsdef mview(self, connection=None):
selfThe materialized view this table belongs to (or None if it's a
real table).
def ismview(self, connection=None):
selfIs this table a materialized view?
def organization(self, connection=None):
selfReturn the organization of this table: either "heap" (for "normal"
tables) or "index" (for index organized tables).
def iternames(cls, connection, owner=<object object at 0xf67a2c80>):
clsdef itercolumns(self, connection=None):
selfGenerator that yields all column objects of the Table self.
def iterrecords(self, connection=None):
selfGenerator that yields all records of the table self.
def itercomments(self, connection=None):
selfGenerator that yields all column comments of the table self.
def _iterconstraints(self, connection, cond):
selfdef iterconstraints(self, connection=None):
selfGenerator that yields all constraints for this table.
def pk(self, connection=None):
selfReturn the primary key constraint for this table (or None if the
table has no primary key constraint).
def iterreferences(self, connection=None):
selfdef iterreferencedby(self, connection=None):
selfclass Constraint(Object):
Base class of all constraints (primary key constraints, foreign key constraints and unique constraints).
def cdate(self, connection=None):
selfdef udate(self, connection=None):
selfdef iternames(cls, connection, owner=<object object at 0xf67a2c80>):
clsdef fixname(cls, code):
clsclass PrimaryKey(Constraint):
Models a primary key constraint in the database.
def itercolumns(self, connection=None):
selfReturn an iterator over the columns this primary key consists of.
def createddl(self, connection=None, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef iterreferencedby(self, connection=None):
selfdef iterreferences(self, connection=None):
selfdef table(self, connection=None):
selfReturn the Table self belongs to.
class Comment(Object):
Models a column comment in the database.
def createddl(self, connection=None, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef fixname(cls, code):
clsdef cdate(self, connection=None):
selfdef udate(self, connection=None):
selfdef iterreferences(self, connection=None):
selfdef iterreferencedby(self, connection=None):
selfclass ForeignKey(Constraint):
Models a foreign key constraint in the database.
def createddl(self, connection=None, term=True):
selfdef _ddl(self, connection, cmd, term):
selfdef dropddl(self, connection=None, term=True):
selfdef enableddl(self, connection=None, term=True):
selfdef disableddl(self, connection=None, term=True):
selfdef iterreferencedby(self, connection=None):
selfdef iterreferences(self, connection=None):
selfdef table(self, connection=None):
selfReturn the Table self belongs to.
def pk(self, connection=None):
selfReturn the primary key referenced by self.
def itercolumns(self, connection=None):
selfReturn an iterator over the columns this foreign key consists of.
def isenabled(self, connection=None):
selfReturn whether this constraint is enabled.
class Index(MixinNormalDates, Object):
Models an index in the database.
def createddl(self, connection=None, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef iternames(cls, connection, owner=<object object at 0xf67a2c80>):
clsdef fixname(cls, code):
clsdef constraint(self, connection=None):
selfIf this index is generated by a constraint, return the constraint
otherwise return None.
def isconstraint(self, connection=None):
selfIs this index generated by a constraint?
def iterreferences(self, connection=None):
selfdef table(self, connection=None):
selfReturn the Table self belongs to.
class UniqueConstraint(Constraint):
Models a unique constraint in the database.
def createddl(self, connection=None, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef iterreferencedby(self, connection=None):
selfdef iterreferences(self, connection=None):
selfdef table(self, connection=None):
selfReturn the Table self belongs to.
class Synonym(Object):
Models a synonym in the database.
def createddl(self, connection=None, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef fixname(cls, code):
clsdef cdate(self, connection=None):
selfdef udate(self, connection=None):
selfdef iterreferences(self, connection=None, done=None):
selfdef getobject(self, connection=None):
selfGet the object for which self is a synonym.
class View(MixinNormalDates, Object):
Models a view in the database.
def createddl(self, connection=None, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef fixname(cls, code):
clsdef iterrecords(self, connection=None):
selfclass MaterializedView(View):
Models a meterialized view in the database.
def createddl(self, connection=None, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef fixname(cls, code):
clsdef iterreferences(self, connection=None):
selfdef iterreferencedby(self, connection=None):
selfclass Library(Object):
Models a library in the database.
def createddl(self, connection=None, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef fixname(cls, code):
clsclass Argument(object):
Argument objects hold information about the arguments of a
stored procedure.
def __init__(self, name, position, datatype, isin, isout):
selfdef __repr__(self):
selfclass Callable(MixinNormalDates, MixinCodeDDL, Object):
Models a callable object in the database, i.e. functions and procedures.
def __init__(self, name, owner=None, connection=None):
selfdef _calcargs(self, cursor):
selfdef _getargs(self, cursor, *args, **kwargs):
selfdef _wraparg(self, cursor, arginfo, arg):
selfdef _unwraparg(self, arginfo, cursor, value):
selfdef _makerecord(self, cursor, args):
selfdef iterarguments(self, connection=None):
selfGenerator that yields all arguments of the function/procedure self.
class Procedure(Callable):
Models a procedure in the database. A Procedure object can be
used as a wrapper for calling the procedure with keyword arguments.
def __call__(self, cursor, *args, **kwargs):
selfCall the procedure with arguments args and keyword arguments
kwargs. cursor must be a ll.orasql cursor. This will
return a Record object containing the result of the call (i.e.
this record will contain all specified and all out parameters).
class Function(Callable):
Models a function in the database. A Function object can be
used as a wrapper for calling the function with keyword arguments.
def __call__(self, cursor, *args, **kwargs):
selfCall the function with arguments args and keyword arguments
kwargs. cursor must be an ll.orasql cursor.
This will return a tuple containing the result and a Record
object containing the modified parameters (i.e. this record will contain
all specified and out parameters).
class Package(MixinNormalDates, MixinCodeDDL, Object):
Models a package in the database.
class PackageBody(MixinNormalDates, MixinCodeDDL, Object):
Models a package body in the database.
class Type(MixinNormalDates, MixinCodeDDL, Object):
Models a type definition in the database.
class Trigger(MixinNormalDates, MixinCodeDDL, Object):
Models a trigger in the database.
class JavaSource(MixinNormalDates, Object):
Models Java source code in the database.
def createddl(self, connection=None, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef fixname(cls, code):
clsclass Privilege(object):
Models a database object privilege (i.e. a grant).
A Privilege object has the following attributes:
privilege(string)The type of the privilege (
EXECUTEetc.)name(string)The name of the object for which this privilege grants access
owner(string orNone)the owner of the object
grantor(string orNone)Who granted this privilege?
grantee(string orNone)To whom has this privilege been granted?
connection(ConnectionorNone)The database connection
def __init__(self, privilege, name, grantor, grantee, owner=None, connection=None):
selfdef __repr__(self):
selfdef __str__(self):
selfdef getconnection(self, connection):
selfdef getcursor(self, connection):
selfproperty connectstring:
def __get__(self):
def iterobjects(cls, connection, owner=<object object at 0xf67a2c80>):
clsGenerator that yields object privileges. For the meaning of owner
see Object.iternames.
def grantddl(self, connection=None, term=True, mapgrantee=True):
selfReturn SQL code to grant this privilege. If mapgrantee is a list
or a dictionary and self.grantee is not in this list (or dictionary)
no command will be returned. If it's a dictionary and self.grantee is
in it, the privilege will be granted to the user specified as the value
instead of the original one. If mapgrantee is true (the default)
the privilege will be granted to the original grantee.
class Column(Object):
Models a single column of a table in the database. This is used to output
ALTER TABLE ... statements for adding, dropping and modifying columns.
def _getcolumnrecord(self, cursor):
selfdef addddl(self, connection=None, term=True):
selfdef modifyddl(self, connection, cursorold, cursornew, term=True):
selfdef dropddl(self, connection=None, term=True):
selfdef table(self):
selfdef cdate(self, connection=None):
selfdef udate(self, connection=None):
selfdef iterreferences(self, connection=None):
selfdef iterreferencedby(self, connection=None):
selfdef datatype(self, connection=None):
selfThe SQL type of this column.
def default(self, connection=None):
selfThe SQL default value for this column.
def nullable(self, connection=None):
selfIs this column nullable?
def comment(self, connection=None):
selfThe comment for this column.
class User(object):
Models a user in the database
def __init__(self, name, connection=None):
selfdef __repr__(self):
selfdef __str__(self):
selfdef __eq__(self, other):
selfdef __ne__(self, other):
selfdef __hash__(self):
selfdef iternames(cls, connection):
clsGenerator that yields the names of all users
def iterobjects(cls, connection):
clsGenerator that yields all user objects.
class OracleConnection(ll.url.Connection):
def __init__(self, context, connection, mode):
selfdef open(self, url, mode='rb'):
selfdef close(self):
selfdef _type(self, url):
selfdef _infofromurl(self, url):
selfdef _objectfromurl(self, url):
selfdef isdir(self, url):
selfdef isfile(self, url):
selfdef mimetype(self, url):
selfdef owner(self, url):
selfdef cdate(self, url):
selfdef mdate(self, url):
selfdef _listdir(self, url, pattern=None, files=True, dirs=True):
selfdef listdir(self, url, pattern=None):
selfdef files(self, url, pattern=None):
selfdef dirs(self, url, pattern=None):
selfdef __repr__(self):
selfclass OracleFileResource(ll.url.Resource):
An OracleFileResource wraps an Oracle database object (like a
table, view, function, procedure etc.) in a file-like API.
def __init__(self, connection, url, mode='rb'):
selfdef read(self, size=-1):
selfdef write(self, data):
selfdef mimetype(self):
selfdef cdate(self):
selfdef mdate(self):
selfdef close(self):
selfclass OracleSchemeDefinition(ll.url.SchemeDefinition):
def _connect(self, url, context=None, **kwargs):
selfdef open(self, url, mode='rb', context=None):
selfdef closeall(self, context):
self