SQLObject¶

Ian Bicking <ianb@colorstudy.com> · $LastChangedDate$

pip install -U SQLObject

easy_install -U SQLObject

easy_install SQLObject==dev

easy_install SQLObject==bugfix

setup(...
  install_requires=["SQLObject==bugfix,>=0.7.1dev-r1485"],
)

mysql:///localhost/test?charset=latin1

>>> from sqlobject import *

scheme://[user[:password]@]host[:port]/database[?parameters]

mysql://user:password@host/database
mysql://host/database?debug=1
postgres://user@host/database?debug=&cache=
postgres:///full/path/to/socket/database
postgres://host:5432/database
sqlite:///full/path/to/database
sqlite:/C:/full/path/to/database
sqlite:/:memory:

>>> import os
>>> db_filename = os.path.abspath('data.db')
>>> connection_string = 'sqlite:' + db_filename
>>> connection = connectionForURI(connection_string)
>>> sqlhub.processConnection = connection

>>> class Person(SQLObject):
...
...     firstName = StringCol()
...     middleInitial = StringCol(length=1, default=None)
...     lastName = StringCol()

CREATE TABLE person (
    id INT PRIMARY KEY AUTO_INCREMENT,
    first_name TEXT,
    middle_initial CHAR(1),
    last_name TEXT
);

>>> Person.createTable()
[]

>>> Person(firstName="John", lastName="Doe")
<Person 1 firstName='John' middleInitial=None lastName='Doe'>

>>> Person.get(1)
<Person 1 firstName='John' middleInitial=None lastName='Doe'>

>>> p = Person.get(1)
>>> p
<Person 1 firstName='John' middleInitial=None lastName='Doe'>
>>> p.firstName
'John'
>>> p.middleInitial = 'Q'
>>> p.middleInitial
'Q'
>>> p2 = Person.get(1)
>>> p2
<Person 1 firstName='John' middleInitial='Q' lastName='Doe'>
>>> p is p2
True

>>> # This will make SQLObject print out the SQL it executes:
>>> Person._connection.debug = True
>>> p = Person(firstName='Bob', lastName='Hope')
 1/QueryIns:  INSERT INTO person (first_name, middle_initial, last_name) VALUES ('Bob', NULL, 'Hope')
 1/QueryR  :  INSERT INTO person (first_name, middle_initial, last_name) VALUES ('Bob', NULL, 'Hope')
 1/COMMIT  :  auto
 1/QueryOne:  SELECT first_name, middle_initial, last_name FROM person WHERE ((person.id) = (2))
 1/QueryR  :  SELECT first_name, middle_initial, last_name FROM person WHERE ((person.id) = (2))
 1/COMMIT  :  auto
>>> p
<Person 2 firstName='Bob' middleInitial=None lastName='Hope'>
>>> p.middleInitial = 'Q'
 1/Query   :  UPDATE person SET middle_initial = ('Q') WHERE id = (2)
 1/QueryR  :  UPDATE person SET middle_initial = ('Q') WHERE id = (2)
 1/COMMIT  :  auto
>>> p2 = Person.get(1)
>>> # Note: no database access, since we're just grabbing the same
>>> # instance we already had.

>>> p.set(firstName='Robert', lastName='Hope Jr.')

>>> Person._connection.debug = True
>>> peeps = Person.select(Person.q.firstName=="John")
>>> list(peeps)
 1/Select  :  SELECT person.id, person.first_name, person.middle_initial, person.last_name FROM person WHERE ((person.first_name) = ('John'))
 1/QueryR  :  SELECT person.id, person.first_name, person.middle_initial, person.last_name FROM person WHERE ((person.first_name) = ('John'))
 1/COMMIT  :  auto
[<Person 1 firstName='John' middleInitial='Q' lastName='Doe'>]

>>> peeps = Person.select(
...         OR(Person.q.firstName == "John",
...            LIKE(Person.q.lastName, "%Hope%")))
>>> list(peeps)
 1/Select  :  SELECT person.id, person.first_name, person.middle_initial, person.last_name FROM person WHERE (((person.first_name) = ('John')) OR (person.last_name LIKE ('%Hope%')))
 1/QueryR  :  SELECT person.id, person.first_name, person.middle_initial, person.last_name FROM person WHERE (((person.first_name) = ('John')) OR (person.last_name LIKE ('%Hope%')))
 1/COMMIT  :  auto
[<Person 1 firstName='John' middleInitial='Q' lastName='Doe'>, <Person 2 firstName='Robert' middleInitial='Q' lastName='Hope Jr.'>]

>>> Person._connection.debug = False  # Need for doctests
>>> peeps = Person.select("""person.first_name = 'John' AND
...                          person.last_name LIKE 'D%'""")

>>> peeps = Person.select(Person.q.firstName=="John")

>>> peeps = Person.selectBy(firstName="John", lastName="Doe")

>>> class Address(SQLObject):
...
...     street = StringCol()
...     city = StringCol()
...     state = StringCol(length=2)
...     zip = StringCol(length=9)
...     person = ForeignKey('Person')
>>> Address.createTable()
[]

class Person(SQLObject):
    ...
    addresses = MultipleJoin('Address')

>>> Person.sqlmeta.addJoin(MultipleJoin('Address',
...                        joinMethodName='addresses'))

>>> p.addresses
[]
>>> Address(street='123 W Main St', city='Smallsville',
...         state='MN', zip='55407', person=p)
<Address 1 ...>
>>> p.addresses
[<Address 1 ...>]

>>> class User(SQLObject):
...
...     class sqlmeta:
...         # user is a reserved word in some databases, so we won't
...         # use that for the table name:
...         table = "user_table"
...
...     username = StringCol(alternateID=True, length=20)
...     # We'd probably define more attributes, but we'll leave
...     # that exercise to the reader...
...
...     roles = RelatedJoin('Role')

>>> class Role(SQLObject):
...
...     name = StringCol(alternateID=True, length=20)
...
...     users = RelatedJoin('User')

>>> User.createTable()
[]
>>> Role.createTable()
[]

>>> bob = User(username='bob')
>>> tim = User(username='tim')
>>> jay = User(username='jay')
>>> admin = Role(name='admin')
>>> editor = Role(name='editor')
>>> bob.addRole(admin)
>>> bob.addRole(editor)
>>> tim.addRole(editor)
>>> bob.roles
[<Role 1 name='admin'>, <Role 2 name='editor'>]
>>> tim.roles
[<Role 2 name='editor'>]
>>> jay.roles
[]
>>> admin.users
[<User 1 username='bob'>]
>>> editor.users
[<User 1 username='bob'>, <User 2 username='tim'>]

>>> User.byUsername('bob')
<User 1 username='bob'>
>>> Role.byName('admin')
<Role 1 name='admin'>

>>> Person._connection.debug = False # Needed for doctests
>>> peeps = Person.select(
...         AND(Address.q.personID == Person.q.id,
...             Address.q.zip.startswith('504')))
>>> list(peeps)
[]
>>> peeps = Person.select(
...         AND(Address.q.personID == Person.q.id,
...             Address.q.zip.startswith('554')))
>>> list(peeps)
[<Person 2 firstName='Robert' middleInitial='Q' lastName='Hope Jr.'>]

>>> Person._connection.debug = False  # Needed for doctests
>>> peeps = Person.select("""address.person_id = person.id AND
...                          address.zip LIKE '504%'""",
...                       clauseTables=['address'])

class MyClass(SQLObject):

    class sqlmeta:
        lazyUpdate = True
        cacheValues = False

    columnA = StringCol()
    columnB = IntCol()

    def _set_attr1(self, value):
        # do something with value

    def _get_attr1(self):
        # do something to retrieve value

class Person(SQLObject):
    # ...

    def imageFilename(self):
        return 'images/person-%s.jpg' % self.id

    def _get_image(self):
        if not os.path.exists(self.imageFilename()):
            return None
        f = open(self.imageFilename())
        v = f.read()
        f.close()
        return v

    def _set_image(self, value):
        # assume we get a string for the image
        f = open(self.imageFilename(), 'w')
        f.write(value)
        f.close()

    def _del_image(self, value):
        # We usually wouldn't include a method like this, but for
        # instructional purposes...
        os.unlink(self.imageFilename())

class Person(SQLObject):
    lastName = StringCol()
    firstName = StringCol()

    def _set_lastName(self, value):
        self.notifyLastNameChange(value)
        self._SO_set_lastName(value)

import re

class PhoneNumber(SQLObject):
    phoneNumber = StringCol(length=30)

    _garbageCharactersRE = re.compile(r'[\-\.\(\) ]')
    _phoneNumberRE = re.compile(r'^[0-9]+$')
    def _set_phoneNumber(self, value):
        value = self._garbageCharactersRE.sub('', value)
        if not len(value) >= 10:
            raise ValueError(
                'Phone numbers must be at least 10 digits long')
        if not self._phoneNumberRE.match(value):
            raise ValueError, 'Phone numbers can contain only digits'
        self._SO_set_phoneNumber(value)

    def _get_phoneNumber(self):
        value = self._SO_get_phoneNumber()
        number = '(%s) %s-%s' % (value[0:3], value[3:6], value[6:10])
        if len(value) > 10:
            number += ' ext.%s' % value[10:]
        return number

MyTable.select(u'value' == MyTable.q.name)
MyTable.select(MyTable.q.name != u'value')
MyTable.select(OR(MyTable.q.col1 == u'value1', MyTable.q.col2 != u'value2'))
MyTable.selectBy(name = u'value')
MyTable.selectBy(col1=u'value1', col2=u'value2')
MyTable.byCol1(u'value1') # if col1 is an alternateID

MyTable.select((MyTable.q.name + MyTable.q.surname) == u'value')

MyTable.select((MyTable.q.name + MyTable.q.surname) == u'value'.encode(dbEncoding))

CREATE TABLE person (
    id SERIAL,
    username VARCHAR(100) NOT NULL UNIQUE
);

CREATE TABLE role (
    id SERIAL,
    name VARCHAR(50) NOT NULL UNIQUE
);

CREATE TABLE assigned_roles (
    person INT NOT NULL,
    role INT NOT NULL
);

class Person(SQLObject):
    username = StringCol(length=100, alternateID=True)
    roles = RelatedJoin('Role', joinColumn='person', otherColumn='role',
                        intermediateTable='assigned_roles')
class Role(SQLObject):
    name = StringCol(length=50, alternateID=True)
    roles = RelatedJoin('Person', joinColumn='role', otherColumn='person',
                        intermediateTable='assigned_roles')

conn = DBConnection.PostgresConnection('yada')
trans = conn.transaction()
p = Person.get(1, trans)
p.firstName = 'Bob'
trans.commit()
p.firstName = 'Billy'
trans.rollback()

Person.select(Person.q.name=="value", forUpdate=True, connection=trans)

firstLastIndex = DatabaseIndex('firstName', 'lastName')

lastNameIndex = DatabaseIndex({'expression': 'lower(last_name)'})

lastNameIndex = DatabaseIndex({'column': lastName, 'length': 10})

class Person(SQLObject):
    class sqlmeta:
        fromDatabase = True

class Person(SQLObject):
    class sqlmeta:
        style = MixedCaseStyle(longID=True)
    firstName = StringCol()
    lastName = StringCol()

CREATE TABLE Person (
    PersonID INT PRIMARY KEY,
    FirstName Text,
    LastName Text
)

__connection__.style = MixedCaseStyle(longID=True)

import logging
logging.basicConfig(
    filename='test.log',
    format='[%(asctime)s] %(name)s %(levelname)s: %(message)s',
    level=logging.DEBUG,
)
log = logging.getLogger("TEST")
log.info("Log started")

__connection__ = "sqlite:/:memory:?debug=1&logger=TEST&loglevel=debug"

import time
import sqlobject
import MySQLdb.converters

def _mysql_timestamp_converter(raw):
         """Convert a MySQL TIMESTAMP to a floating point number representing
         the seconds since the Un*x Epoch. It uses custom code the input seems
         to be the new (MySQL 4.1+) timestamp format, otherwise code from the
         MySQLdb module is used."""
         if raw[4] == '-':
             return time.mktime(time.strptime(raw, '%Y-%m-%d %H:%M:%S'))
         else:
             return MySQLdb.converters.mysql_timestamp_converter(raw)

conversions = MySQLdb.converters.conversions.copy()
conversions[MySQLdb.constants.FIELD_TYPE.TIMESTAMP] = _mysql_timestamp_converter

MySQLConnection = sqlobject.mysql.builder()
connection = MySQLConnection(user='foo', db='somedb', conv=conversions)

# fbdb1 is on a Windows server:
fbdb1 = c:\Firebird\sample\Employee.fdb

from sqlobject.events import listen, RowUpdateSignal, RowCreatedSignal
from model import Users

def update_listener(instance, kwargs):
    """keep "last_updated" field current"""
    import datetime
    # BAD method 1, causes infinite recursion?
    # instance should be read-only
    instance.last_updated = datetime.datetime.now()
    # GOOD method 2
    kwargs['last_updated'] = datetime.datetime.now()

def created_listener(instance, kwargs, post_funcs):
    """"email me when new users added"""
    # email() implementation left as an exercise for the reader
    msg = "%s just was just added to the database!" % kwargs['name']
    email(msg)

listen(update_listener, Users, RowUpdateSignal)
listen(created_listener, Users, RowCreatedSignal)

from sqlobject.sqlbuilder import LEFTJOINOn
MyTable.select(
    join=LEFTJOINOn(Table1, Table2,
                    Table1.q.name == Table2.q.value))

SELECT my_table.* FROM my_table, table1
LEFT JOIN table2 ON table1.name = table2.value;

MyTable.select(
    join=LEFTJOINOn(None, Table1,
                    MyTable.q.name == Table1.q.value))

SELECT my_table.* FROM my_table
LEFT JOIN table2 ON my_table.name = table1.value;

from sqlobject.sqlbuilder import Alias
alias = Alias(MyTable, "my_table_alias")
MyTable.select(MyTable.q.name == alias.q.value)

SELECT my_table.* FROM my_table, my_table AS my_table_alias
WHERE my_table.name = my_table_alias.value;

from sqlobject.sqlbuilder import LEFTJOINOn, Alias
alias = Alias(OtherTable, "other_table_alias")
MyTable.select(MyTable.q.name == OtherTable.q.value,
    join=LEFTJOINOn(MyTable, alias, MyTable.col1 == alias.q.col2))

SELECT my_table.* FROM other_table,
    my_table LEFT JOIN other_table AS other_table_alias
WHERE my_table.name == other_table.value AND
    my_table.col1 = other_table_alias.col2.

from sqlobject.sqlbuilder import EXISTS, Select
select = Test1.select(EXISTS(Select(Test2.q.col2, where=(Outer(Test1).q.col1 == Test2.q.col2))))

SELECT test1.id, test1.col1 FROM test1 WHERE
EXISTS (SELECT test2.col2 FROM test2 WHERE (test1.col1 = test2.col2))

SELECT customer.id, customer.first_name, customer.last_name,
       contact.id, contact.address
FROM customer
LEFT JOIN contact ON contact.customer_id = customer.id

for customer in Customer.select():
    print customer.firstName, customer.lastName
    for contact in customer.contacts:
        print '   ', contact.phoneNumber

custContacts = {}
for contact in Contact.select():
    custContacts.setdefault(contact.customerID, []).append(contact)
for customer in Customer.select():
    print customer.firstName, customer.lastName
    for contact in custContacts.get(customer.id, []):
        print '   ', contact.phoneNumber

query = Customer.q.firstName.startswith('J')
custContacts = {}
for contact in Contact.select(AND(Contact.q.customerID == Customer.q.id,
                                  query)):
    custContacts.setdefault(contact.customerID, []).append(contact)
for customer in Customer.select(query):
    print customer.firstName, customer.lastName
    for contact in custContacts.get(customer.id, []):
        print '   ', contact.phoneNumber

>>> class User(SQLObject):
...     class sqlmeta:
...         table = "user_table"
...     username = StringCol(alternateID=True, length=20)
...     roles = SQLRelatedJoin('Role',
...         intermediateTable='user_roles',
...         createRelatedTable=False)

>>> class Role(SQLObject):
...     name = StringCol(alternateID=True, length=20)
...     users = SQLRelatedJoin('User',
...         intermediateTable='user_roles',
...         createRelatedTable=False)

>>> class UserRoles(SQLObject):
...     class sqlmeta:
...         table = "user_roles"
...     user = ForeignKey('User', notNull=True, cascade=True)
...     role = ForeignKey('Role', notNull=True, cascade=True)
...     active = BoolCol(notNull=True, default=False)
...     unique = index.DatabaseIndex(user, role, unique=True)

class SiteSQLObject(SQLObject):
    _connection = DBConnection.MySQLConnection(user='test', db='test')
    _style = MixedCaseStyle()

    # And maybe you want a list of the columns, to autogenerate
    # forms from:
    def columns(self):
        return [col.name for col in self._columns]

class Person(SQLObject):
    firstName = StringCol()
    lastName = StringCol()

class Employee(Person):
    position = StringCol()

CREATE TABLE person (
    id INT PRIMARY KEY,
    first_name TEXT,
    last_name TEXT
);

CREATE TABLE employee (
    id INT PRIMARY KEY
    first_name TEXT,
    last_name TEXT,
    position TEXT
)

class Person(SQLObject):
    firstName = StringCol()
    lastName = StringCol()

    def _get_employee(self):
        value = Employee.selectBy(person=self)
        if value:
            return value[0]
        else:
            raise AttributeError, '%r is not an employee' % self
    def _get_isEmployee(self):
        value = Employee.selectBy(person=self)
        # turn into a bool:
        return not not value
    def _set_isEmployee(self, value):
        if value:
            # Make sure we are an employee...
            if not self.isEmployee:
                Empoyee.new(person=self, position=None)
        else:
            if self.isEmployee:
                self.employee.destroySelf()
    def _get_position(self):
        return self.employee.position
    def _set_position(self, value):
        self.employee.position = value

class Employee(SQLObject):
    person = ForeignKey('Person')
    position = StringCol()

CREATE TABLE invoice_item (
    id INT PRIMARY KEY,
    amount NUMERIC(10, 2),
    currency CHAR(3)
);

class InvoiceItem(SQLObject):
    amount = Currency()
    currency = StringChar(length=3)

    def _get_price(self):
        return Price(self.amount, self.currency)
    def _set_price(self, price):
        self.amount = price.amount
        self.currency = price.currency

class Price(object):
    def __init__(self, amount, currency):
        self._amount = amount
        self._currency = currency

    def _get_amount(self):
        return self._amount
    amount = property(_get_amount)

    def _get_currency(self):
        return self._currency
    currency = property(_get_currency)

    def __repr__(self):
        return '<Price: %s %s>' % (self.amount, self.currency)

class Address(SQLObject):
    street = StringCol()
    city = StringCol()
    state = StringCol(length=2)

    latitude = FloatCol()
    longitude = FloatCol()

    def _init(self, id):
        SQLObject._init(self, id)
        self._coords = SOCoords(self)

    def _get_coords(self):
        return self._coords

class SOCoords(object):
    def __init__(self, so):
        self._so = so

    def _get_latitude(self):
        return self._so.latitude
    def _set_latitude(self, value):
        self._so.latitude = value
    latitude = property(_get_latitude, set_latitude)

    def _get_longitude(self):
        return self._so.longitude
    def _set_longitude(self, value):
        self._so.longitude = value
    longitude = property(_get_longitude, set_longitude)

color = Something(id="blue", r=0, b=100, g=0)

def Color(SQLObject):
    def _init(self, id, connection=None):
        id = str(id)
        SQLObject._init(self, id, connection)

class Image(SQLObject):

    data = StringCol()
    height = IntCol()
    width = IntCol()

    def _set_data(self, value):
        self._SO_set_data(value.encode('base64'))

    def _get_data(self, value):
        return self._SO_get_data().decode('base64')

class Person(SQLObject):
    role = ForeignKey("Role")

class Role(SQLObject):
    person = ForeignKey("Person")

Person.createTable()
Role.createTable()

class Person(SQLObject):
    role = ForeignKey("Role")

class Role(SQLObject):
    person = ForeignKey("Person")

constraints = Person.createTable(applyConstraints=False)
constraints += Role.createTable(applyConstraints=False)

for constraint in constraints:
    connection.query(constraint)

class MyTable(SQLObject):
    ...

my_table_row = MyTable.get(id)

SELECT my_column, COUNT(*) FROM my_table GROUP BY my_column;

class SomeObject(SQLObject):
    class sqlmeta:
        createSQL = { 'mysql' : 'ALTER TABLE some_object ENGINE InnoDB' }
    # your columns here

class SomeObject(SQLObject):
    class sqlmeta:
        createSQL = { 'mysql' : [
            'ALTER TABLE some_object ENGINE InnoDB',
            '''ALTER TABLE some_object CHARACTER SET utf8
                COLLATE utf8_estonian_ci''']
            }

>>> from sqlobject.sqlbuilder import *
>>> person = table.person
# person is now equivalent to the Person.q object from the SQLObject
# documentation
>>> person
person
>>> person.first_name
person.first_name
>>> person.first_name == 'John'
person.first_name = 'John'
>>> name = 'John'
>>> person.first_name != name
person.first_name != 'John'
>>> AND(person.first_name == 'John', person.last_name == 'Doe')
(person.first_name = 'John' AND person.last_name = 'Doe')

# This isn't what you want:
>> person.first_name == 'John' & person.last_name == 'Doe'
(person.first_name = ('John' AND person.last_name)) = 'Doe')
# This is:
>> (person.first_name == 'John') & (person.last_name == 'Doe')
((person.first_name = 'John') AND (person.last_name == 'Doe'))

>> person.created < func.NOW()

>>> from sqlobject.sqlbuilder import *
>> select = Select(['name', 'AVG(salary)'], staticTables=['employees'],
>>     groupBy='name') # create an instance
>> query = connection.sqlrepr(select) # Convert to SQL string:
>>     # SELECT name, AVG(salary) FROM employees GROUP BY name
>> rows = connection.queryAll(query) # Execute the query
>>     # and get back the results as a list of rows
>>     # where every row is a sequence of length 2 (name and average salary)

>> insert = Insert('person', valueList=[('name', 'Test'), ('age', 42)])
       # or
>> insert = Insert('person', valueList=[{'name': 'Test'}, {'age': 42}])
>> query = connection.sqlrepr(insert)
       # Both generate the same query:
       # INSERT INTO person (name, age) VALUES ('Test', 42)
>> connection.query(query)

>> insert = Insert('person', values={'name': 'Test', 'age': 42})
>> query = connection.sqlrepr(insert)
       # The query is the same
       # INSERT INTO person (name, age) VALUES ('Test', 42)
>> connection.query(query)

>> update = Update('person',
>>     values={'name': 'Test', 'age': 42}, where='id=1')
>> query = connection.sqlrepr(update)
       # UPDATE person SET name='Test', age=42 WHERE id=1
>> connection.query(query)

>> update = Delete('person', where='id=1')
>> query = connection.sqlrepr(update)
       # DELETE FROM person WHERE id=1
>> connection.query(query)

>> select1 = Select(['min', func.MIN(const.salary)], staticTables=['employees'])
>> select2 = Select(['max', func.MAX(const.salary)], staticTables=['employees'])
>> union = Union(select1, select2)
>> query = connection.sqlrepr(union)
       # SELECT 'min', MIN(salary) FROM employees
       #    UNION
       # SELECT 'max', MAX(salary) FROM employees
>> rows = connection.queryAll(query)

>> PersonWorkplace.deleteMany(where=
   ((PersonWorkplace.q.WorkplaceID==Workplace.q.id) &
   (Workplace.q.id==SOME_ID)))

>> PersonWorkplace.deleteMany(where=
   IN(PersonWorkplace.q.WorkplaceID,
   Select(Workplace.q.id, Workplace.q.id==SOME_ID)))

start = 20
size = 10
query = Table.select()
results = query[start:start+size]
total = query.count()
print "Showing page %i of %i" % (start/size + 1, total/size + 1)

Schools.select(Schools.q.student_satisfaction>90).throughTo.teachers

Person.select(Person.q.name=='Steve').throughTo.father.throughTo.father

class MyTable(SQLObject):
    class sqlmeta:
        createSQL = {'postgres': [
            "ALTER TABLE my_table ADD CHECK(my_field != '');",
            ]}
    myField = StringCol()

class Person(SQLObject):
    name = StringCol()
    age = FloatCol()

class Role(InheritableSQLObject):
    department = StringCol()

class Student(Role):
    year = IntCol()

class Professor(Role):
    timetable = StringCol()

class Person(SQLObject):
    name = StringCol()
    age = FloatCol()
    role = ForeignKey("Role")

from sqlobject import *
from sqlobject.inheritance import InheritableSQLObject

__connection__ = "sqlite:/:memory:"

class Role(InheritableSQLObject):
   department = StringCol()

class Student(Role):
   year = IntCol()

class Professor(Role):
   timetable = StringCol(default=None)

class Person(SQLObject):
   name = StringCol()
   age = FloatCol()
   role = ForeignKey("Role", default=None)

Role.createTable()
Student.createTable()
Professor.createTable()
Person.createTable()

first_year = Student(department="CS", year=1)
lecturer = Professor(department="Mathematics")

student = Person(name="A student", age=21, role=first_year)
professor = Person(name="A professor", age=42, role=lecturer)

print student.role
print professor.role

<Student 1 year=1 department='CS'>
<Professor 2 timetable=None department='Mathematics'>

print list(Role.select())

[<Student 1 year=1 department='CS'>, <Professor 2 timetable=None department='Mathematics'>]

class Role(InheritableSQLObject):
    department = StringCol()
    persons = MultipleJoin("Person")

for role in Role.select():
    print role.persons

[<Person 1 name='A student' age=21.0 roleID=1>]
[<Person 2 name='A professor' age=42.0 roleID=2>]

class Role(InheritableSQLObject):
    department = StringCol()
    persons = RelatedJoin("Person")

class Student(Role):
    year = IntCol()

class Professor(Role):
    timetable = StringCol(default=None)

class Person(SQLObject):
    name = StringCol()
    age = FloatCol()
    roles = RelatedJoin("Role")

Role.createTable()
Student.createTable()
Professor.createTable()
Person.createTable()

first_year = Student(department="CS", year=1)
lecturer = Professor(department="Mathematics")

student = Person(name="A student", age=21)
student.addRole(first_year)
professor = Person(name="A professor", age=42)
professor.addRole(lecturer)

print student.roles
print professor.roles

for role in Role.select():
    print role.persons

[<Student 1 year=1 department='CS'>]
[<Professor 2 timetable=None department='Mathematics'>]
[<Person 1 name='A student' age=21.0>]
[<Person 2 name='A professor' age=42.0>]

from sqlobject.inheritance import InheritableSQLObject
class Person(InheritableSQLObject):
    firstName = StringCol()
    lastName = StringCol()

class Employee(Person):
    _inheritable = False
    position = StringCol()

CREATE TABLE person (
    id INT PRIMARY KEY,
    child_name TEXT,
    first_name TEXT,
    last_name TEXT
);

CREATE TABLE employee (
    id INT PRIMARY KEY,
    position TEXT
)

p = Person(firstName='John', lastName='Doe')
e = Employee(firstName='Jane', lastName='Doe', position='Chief')
p2 = Person.get(1)

*Person*
id
    child_name
    first_name
    last_name
0
    Null
    John
    Doe
1
    Employee
    Jane
    Doe

*Employee*
id
    position
1
    Chief

Employee.select(AND(Employee.q.firstName == 'Jane', Employee.q.position == 'Chief')) will return Jane Doe
Employee.select(AND(Person.q.firstName == 'Jane', Employee.q.position == 'Chief')) will return Jane Doe
Employee.select(Employee.q.lastName == 'Doe') will only return Jane Doe (as Joe isn't an employee)
Person.select(Person.q.lastName == 'Doe') will return both entries.

SELECT employee.id, person.first_name, person.last_name
FROM person, employee WHERE person.first_name = 'Jane'
AND employee.position = 'Chief' AND person.id = employee.id

class MyClass(SQLObject):
    name = StringCol()
    versions = Versioning()

mc = MyClass(name='fleem')

mc.set(name='morx')
assert mc.versions[0].name == 'fleem'

mc.versions[0].restore()
assert mc.name == "fleem"

class Base(InheritableSQLObject):
    name = StringCol()
    versions = Versioning()

class Child(Base):
    toy = StringCol()

class PhoneNumber(SQLObject):
  number = StringCol()
  calls = SQLMultipleJoin('PhoneCall')
  incoming = SQLMultipleJoin('PhoneCall', joinColumn='toID')

class PhoneCall(SQLObject):
  phoneNumber = ForeignKey('PhoneNumber')
  to = ForeignKey('PhoneNumber')
  minutes = IntCol()

class ViewPhoneCall(ViewSQLObject):
  class sqlmeta:
      idName = PhoneCall.q.id
      clause = PhoneCall.q.phoneNumberID==PhoneNumber.q.id

  minutes = IntCol(dbName=PhoneCall.q.minutes)
  number = StringCol(dbName=PhoneNumber.q.number)
  phoneNumber = ForeignKey('PhoneNumber', dbName=PhoneNumber.q.id)
  call = ForeignKey('PhoneCall', dbName=PhoneCall.q.id)

class ViewPhone(ViewSQLObject):
  class sqlmeta:
      idName = PhoneNumber.q.id
      clause = PhoneCall.q.phoneNumberID==PhoneNumber.q.id

  minutes = IntCol(dbName=func.SUM(PhoneCall.q.minutes))
  numberOfCalls = IntCol(dbName=func.COUNT(PhoneCall.q.phoneNumberID))
  number = StringCol(dbName=PhoneNumber.q.number)
  phoneNumber = ForeignKey('PhoneNumber', dbName=PhoneNumber.q.id)
  calls = SQLMultipleJoin('PhoneCall', joinColumn='phoneNumberID')
  vCalls = SQLMultipleJoin('ViewPhoneCall', joinColumn='phoneNumberID')

$ git clone git://github.com/formencode/formencode.git
$ cd formencode
$ sudo python setup.py develop

$ git clone git clone git://github.com/sqlobject/sqlobject.git
$ cd sqlobject
$ sudo python setup.py develop

assert some_condition(a, b), (
    "Some condition failed, %r isn't right!" % a)

short = 3
longerVar = 4

if x == 4:
    do stuff

func(arg1='a', arg2='b')
func((a + b)*10)

short    =3
longerVar=4

if x==4: do stuff

func(arg1 = 'a', arg2 = 'b')
func(a,b)
func( a, b )
[ 1, 2, 3 ]

class AClass(object):
    """
    doc string...
    """

if isinstance(var, str): ...
# Bad:
if type(var) is StringType: ...

class MyTable(SQLObject):
    class sqlmeta:
        createParamsPre = 'TEMPORARY IF NOT EXISTS'
        createParamsPre = {temporary: True, ifNotExists: True,
                           'postgres': 'LOCAL'}
        createParamsPost = 'ENGINE InnoDB'
        createParamsPost = {'mysql': 'ENGINE InnoDB',
                            'postgres': 'WITH OIDS'}

class MyClass(SQLObject):
    function1 = ExpressionCol(func.my_function(MyClass.q.col1))
    function2 = ExpressionCol('sum(col2)')

>>> from sqlobject import *
>>>
>>> sqlhub.processConnection = connectionForURI('sqlite:/:memory:')
>>>
>>> class Person(SQLObject):
...     fname = StringCol()
...     mi = StringCol(length=1, default=None)
...     lname = StringCol()
...
>>> Person.createTable()

>>> p = Person(fname="John", lname="Doe")
>>> p
<Person 1 fname='John' mi=None lname='Doe'>
>>> p.fname
'John'
>>> p.mi = 'Q'
>>> p2 = Person.get(1)
>>> p2
<Person 1 fname='John' mi='Q' lname='Doe'>
>>> p is p2
True

author:	Ian Bicking <ianb@colorstudy.com>
revision:	$Rev$
date:	$LastChangedDate$

SQLObject¶

Documentation¶

Download SQLObject¶

Repositories¶

SQLObject Community¶

News¶

SQLObject 3.3.0¶

Features¶

Minor features¶

Drivers (work in progress)¶

Documentation¶

Tests¶

SQLObject 3.2.0¶

Minor features¶

Drivers (work in progress)¶

Bug fixes¶

Documentation¶

Tests¶

SQLObject 3.1.0¶

Features¶

Documentation¶

Source code¶

Tests¶

SQLObject 3.0.0¶

Features¶

Minor features¶

Development¶

Documentation¶