SQLAlchemy to ORM (relacyjny maper) baz danych dla pythona, konkurencyjny względem SQLObject.
Object-relational mapping (ORM, O/RM, and O/R mapping) in computer software is a programming technique for converting data between incompatible type systems in object-oriented programming languages. This creates, in effect, a “virtual object database” that can be used from within the programming language. There are both free and commercial packages available that perform object-relational mapping, although some programmers opt to create their own ORM tools...
Definicja przytoczona za – http://en.wikipedia.org/
Obecnie wspiera on następujące bazy poprzez podane pakiety:
Instalacja
Aby zainstalować SQLachemy Można skorzystać z easy_install:
easy_install SQLAlchemy
pip install SQLAlchemy
lub też ściągnąć paczkę i zainstalować standardowo:
python setup.py install
Aby sprawdzić poprawność instalacji uruchamiamy interpreter pythona i wpisujemy.

>> import sqlalchemy >> sqlalchemy.__version__ 0.7.0
Pakiet SQLalchemy została zainstalowany prawidłowo.
Podstawy, Łączenie się z bazą danych
By połączyć się z bazą danych należy zastosować:
from sqlalchemy import *
# postgres
pg_db = create_engine('postgres://scott:tiger@localhost:5432/mydatabase')
# sqlite
sqlite_db = create_engine('sqlite:////absolute/path/to/database.txt')
sqlite_db = create_engine('sqlite:///relative/path/to/database.txt')
sqlite_db = create_engine('sqlite://')
# mysql
mysql_db = create_engine('mysql://localhost/foo')
mysql_db = create_engine('mysql+mysqldb://scott:tiger@localhost/foo')
#oracle
# oracle - cx_oracle
oracle_db = create_engine('oracle://scott:tiger@127.0.0.1:1521/sidname')
# oracle via TNS name
oracle_db = create_engine('oracle+cx_oracle://scott:tiger@tnsname')
By rozpocząć pracę z tabelami trzeba je przypisać do obiektu MetaData. Dopiero wtedy można na nich operować. Oto przykład przypisujący bazę do MetaData oraz tworzący tabelę:
from sqlalchemy import *
#driver://login:haslo@host:port/nazwa_bazy
>>> engine=create_engine('oracle://django:password@xe')
>>> conn=engine.connect()
>>> result=conn.execute('select id, domain, name from django_site')
>>> for row in result:
... print 'id: %s Domena: %s Nazwa: %s' % ( row['id'],row['domain'],row['name
'])
...
id: 1 Domena: example.com Nazwa: example.com
# definicja tabeli 'user_sql'
>>> user_db=Table(
... 'user_sql',metadata,
... Column('id',Integer, primary_key=True),
... Column('user_name', Unicode(16),unique=True, nullable=False),
... Column('email', Unicode(255), unique=True, nullable=False),
... Column('password', Unicode(40), nullable=False),
... Column('first_name', Unicode(255), default=''),
... Column('last_name', Unicode(255), default=''))
# utworzenie tabeli 'user_sql'
user_db.create(engine)
Stworzyliśmy tabelę ‘user_sql’ – składnia SQL:
(SQL wygenerowano za pomocą programu TOAD For Oracle – http://www.quest.com )
ALTER TABLE DJANGO.USER_SQL DROP PRIMARY KEY CASCADE; DROP TABLE DJANGO.USER_SQL CASCADE CONSTRAINTS; CREATE TABLE DJANGO.USER_SQL ( ID INTEGER NOT NULL, USER_NAME NVARCHAR2(16) NOT NULL, EMAIL NVARCHAR2(255) NOT NULL, PASSWORD NVARCHAR2(40) NOT NULL, FIRST_NAME NVARCHAR2(255), LAST_NAME NVARCHAR2(255) ) TABLESPACE USERS PCTUSED 0 PCTFREE 10 INITRANS 1 MAXTRANS 255 STORAGE ( INITIAL 64K NEXT 1M MINEXTENTS 1 MAXEXTENTS UNLIMITED PCTINCREASE 0 BUFFER_POOL DEFAULT ) LOGGING NOCOMPRESS NOCACHE NOPARALLEL MONITORING; ALTER TABLE DJANGO.USER_SQL ADD ( PRIMARY KEY (ID) USING INDEX TABLESPACE USERS PCTFREE 10 INITRANS 2 MAXTRANS 255 STORAGE ( INITIAL 64K NEXT 1M MINEXTENTS 1 MAXEXTENTS UNLIMITED PCTINCREASE 0 ), UNIQUE (USER_NAME) USING INDEX TABLESPACE USERS PCTFREE 10 INITRANS 2 MAXTRANS 255 STORAGE ( INITIAL 64K NEXT 1M MINEXTENTS 1 MAXEXTENTS UNLIMITED PCTINCREASE 0 ), UNIQUE (EMAIL) USING INDEX TABLESPACE USERS PCTFREE 10 INITRANS 2 MAXTRANS 255 STORAGE ( INITIAL 64K NEXT 1M MINEXTENTS 1 MAXEXTENTS UNLIMITED PCTINCREASE 0 ));
users_db to obiekt odnoszący się do naszej tabeli. Jeżeli tabela już istnieje to odwołać możemy się do niej w następujący sposób:
users_db = Table('users_sql', metadata, autoload=True)
Dodawanie danych
By dodać dane do bazy danych możemy skorzystać z dwóch sposobów:
from sqlalchemy import *
#driver://login:haslo@host:port/nazwa_bazy
db = create_engine('oracle://django:password@xe')
# przypisanie tabel do MetaData
metadata = MetaData(engine)
metadata.echo = True
users_db = Table('users_sql', metadata, autoload=True)
# dodawanie wierszy
>>> metadata.echo=True
>>> i=user_db.insert()
>>> print i
# Polecenie SQL
INSERT INTO user_sql (id, user_name, email, password, first_name, last_name) VAL
UES (:id, :user_name, :email, :password, :first_name, :last_name)
>>> metadata.bind=engine
# sposób 1
>>> engine.execute(i,id=3,user_name='edyta123', email='ed@wp.pl', password='foo123')
<sqlalchemy.engine.base.ResultProxy object at 0x02CCBBB0>
>>>
# sposób 2
>>>engine.execute(i,{'id': 5, 'user_name': 'zdzichu', 'email': 'sw@wp.pl', 'password': '42zdzich'},
{'id': 6, 'user_name': 'lepper123', 'email': 'lepp@qo.pl', 'password': '0700LEP'},
{'id': 7, 'user_name': 'Carla', 'email': 'clara@wp.pl', 'password': 'foobar'})
<sqlalchemy.engine.base.ResultProxy object at 0x02D1A550>
Pierwszy sposób – dodaje jeden wiersz bazując na danych przypisany do zmiennych będących nazwami pól tabeli. Drugi zawiera listę słowników i służy do dodawania wielu wpisów. Dodatkowo SQLAlchemy ‘slaszuje’ wszystkie znaczące znaki w dodawanych danych. Tak więc praktycznie eliminuje to ataki SQL injection.
Pobieranie danych
Wygląda równie sprawnie:
>>> user_db=Table('user_sql', metadata, autoload=True)
>>> s=user_db.select()
>>> rs=s.execute()
>>> row=rs.fetchone()
>>> print 'id: ', row[0]
id: 5
>>> print 'user name: ', row['user_name']
user name: zdzichu
>>> print 'password: ', row['password']
password: 42zdzich
Co pobierze jeden wiersz (fetchone()). By pobrać wszystkie należy skorzystać z fetchall()
>>> rs = s.execute() >>> a = rs.fetchall() >>> for row in a: ... print row.user_name + ' - ' + row.password; ... zdzichu - 42zdzich lepper123 - 0700LEP Carla - foobar edyta - foo123 edyta123 - foo123 edyta1234 - for234
Ograniczanie pobieranych danych można wykonać na wiele sposobów, oto przegląd:
from sqlalchemy import *
#driver://login:haslo@host:port/nazwa_bazy
>>>db = create_engine('oracle://Django:password@xe')
# przypisanie tabel do MetaData
>>>metadata = MetaData(db)
>>>metadata.echo = True
>>>users_db = Table('users_sql', metadata, autoload=True)
>>> def run(stmt):
... rs=stmt.execute()
... for row in rs:
... print row
...
>>>
>>> s=user_db.select(user_db.c.user_name=='Edyta')
>>> run(s)
>>> s=user_db.select(user_db.c.user_name=='edyta')
>>> run(s)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None)
>>> s=user_db.select(user_db.c.user_name=='Edyta')
>>> print s
#Zapytanie SQL wysłane do Bazy Danych Oracle
SELECT user_sql.id, user_sql.user_name, user_sql.email, user_sql.password, user_sql.first_name, user_sql.last_name
FROM user_sql
WHERE user_sql.user_name = :user_name_1
>>>
# Zapytanie z prostym warunkiem jednego pola
>>> s=user_db.select(user_db.c.user_name!='Edyta')
>>> run(s)
(5, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None)
(6, u'lepper123', u'lepp@qo.pl', u'0700LEP', None, None)
(7, u'Carla', u'clara@wp.pl', u'foobar', None, None)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None)
(4, u'edyta1234', u'test', u'for234', None, None)
>>>
>>>s = users_db.select(users_db.c.id < 40)
>>>run(s)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None)
(4, u'edyta1234', u'test', u'for234', None, None)
(5, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None)
(6, u'lepper123', u'lepp@qo.pl', u'0700LEP', None, None)
(7, u'Carla', u'clara@wp.pl', u'foobar', None, None)
# dostępne są też funkcje and_, or_ i not_
>>>s = users_db.select(and_(users_db.c.id < 40, users_db.c.user_name != 'Mary'))
>>>run(s)
(5, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None)
(6, u'lepper123', u'lepp@qo.pl', u'0700LEP', None, None)
(7, u'Carla', u'clara@wp.pl', u'foobar', None, None)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None)
(4, u'edyta1234', u'test', u'for234', None, None)
>>>
# Są też inne funkcje jak "like", "startswith", "endswith"
>>> s = user_db.select(user_db.c.user_name.startswith('e'))
>>> run(s)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None)
(4, u'edyta1234', u'test', u'for234', None, None)
>>>
>>>s = user_db.select(user_db.c.user_name.like('%a%'))
>>>run(s)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None)
(4, u'edyta1234', u'test', u'for234', None, None)
(6, u'lepper123', u'lepp@qo.pl', u'0700LEP', None, None)
>>>
>>>s = user_db.select(user_db.c.user_name.endswith('4'))
>>>run(s)
(4, u'edyta1234', u'test', u'for234', None, None)
>>>
# "func" służy do wywoływania funkcji SQL
>>>s = user_db.select(func.substr(user_db.c.user_name, 2, 1) == 'a')
>>>run(s)
(7, u'Carla', u'clara@wp.pl', u'foobar', None, None)
# pobieranie danych z określonych pól
>>>s = select([user_db.c.id, user_db.c.user_name], user_db.c.user_name != 'Carl')
>>>run(s)
(7, u'Carla')
(2, u'edyta')
(3, u'edyta123')
(4, u'edyta1234')
(6, u'lepper123')
(5, u'zdzichu')
# SQLowy count()
>>>s = select([func.count(user_db.c.id)])
>>>run(s)
(6,)
# count(*) czyli liczymy wszystko
s = select([func.count("*")], from_obj=[user_db])
run(s)
(6,)
Łączenie tabel – JOIN
Przykład wykonywania zapytań z łączonymi tabelami.
from sqlalchemy import *
#driver://login:haslo@host:port/nazwa_bazy</pre>
>>>engine = create_engine('oracle://django:password@xe')
>>>engine.echo = True
>>>metadata = MetaData(engine)
>>>emails2 = Table('emails2', metadata,
... Column('email_id', Integer, primary_key=True),
... Column('address', Unicode(20)),
... Column('user_id', Integer, ForeignKey('user_sql.id')),
)
>>>emails2.create(engine)
>>>i = emails.insert()
>>>metadata.bind=engine
>>> engine.execute(i,
... {'email_id':1, 'address': 'mary@example.com', 'user_id': 1},
... {'email_id':2, 'address': 'john@nowhere.net', 'user_id': 2},
... {'email_id':3, 'address': 'john@example.org', 'user_id': 2},
... {'email_id':4, 'address': 'carl@nospam.net', 'user_id': 4},
... )
<sqlalchemy.engine.base.ResultProxy object at 0x02D25FB0>
>>>
>>>def run(stmt):
... rs = stmt.execute()
... for row in rs:
... print row
# pełen join, brak where, zwraca "dużo" wyników
>>>s = select([user_db, emails2])
>>>run(s)
(1, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None, 1, u'mary@example.com', 1)
(1, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None, 2, u'john@nowhere.net', 2)
(1, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None, 3, u'john@example.org', 2)
(1, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None, 4, u'carl@nospam.net', 4)
(5, u'lepper123', u'lepp@qo.pl', u'0700LEP', None, None, 1, u'mary@example.com',1)
(5, u'lepper123', u'lepp@qo.pl', u'0700LEP', None, None, 2, u'john@nowhere.net',2)
(5, u'lepper123', u'lepp@qo.pl', u'0700LEP', None, None, 3, u'john@example.org',2)
(5, u'lepper123', u'lepp@qo.pl', u'0700LEP', None, None, 4, u'carl@nospam.net',4)
(6, u'Carla', u'clara@wp.pl', u'foobar', None, None, 1, u'mary@example.com', 1)
(6, u'Carla', u'clara@wp.pl', u'foobar', None, None, 2, u'john@nowhere.net', 2)
(6, u'Carla', u'clara@wp.pl', u'foobar', None, None, 3, u'john@example.org', 2)
(6, u'Carla', u'clara@wp.pl', u'foobar', None, None, 4, u'carl@nospam.net', 4)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None, 1, u'mary@example.com', 1)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None, 2, u'john@nowhere.net', 2)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None, 3, u'john@example.org', 2)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None, 4, u'carl@nospam.net', 4)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None, 1, u'mary@example.com', 1)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None, 2, u'john@nowhere.net', 2)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None, 3, u'john@example.org', 2)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None, 4, u'carl@nospam.net', 4)
(4, u'edyta1234', u'test', u'for234', None, None, 1, u'mary@example.com', 1)
(4, u'edyta1234', u'test', u'for234', None, None, 2, u'john@nowhere.net', 2)
(4, u'edyta1234', u'test', u'for234', None, None, 3, u'john@example.org', 2)
(4, u'edyta1234', u'test', u'for234', None, None, 4, u'carl@nospam.net', 4)
>>>
# join z określonym warunkiem łączenia
>>>s = select([user_db, emails2], emails2.c.user_id == user_db.c.id)
>>>run(s)
(1, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None, 1, u'mary@example.com', 1)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None, 2, u'john@nowhere.net', 2)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None, 3, u'john@example.org', 2)
(4, u'edyta1234', u'test', u'for234', None, None, 4, u'carl@nospam.net', 4)
>>>
# obiekt join jest inteligentny
# i określi pole łączenia po ForeignKey
s = join(user_db, emails2).select()
run(s)
(1, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None, 1, u'mary@example.c,1)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None, 2, u'john@nowhere.net',2)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None, 3, u'john@example.org',3)
(4, u'edyta1234', u'test', u'for234', None, None, 4, u'carl@nospam.net',4)
>>>
# wszyscy userzy, nawet jak nie mają emaila
# potrzebny "outer" join.
s = outerjoin(user_db, emails2).select()
run(s)
(1, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None, 1, u'mary@example.com', 1)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None, 2, u'john@nowhere.net', 2)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None, 3, u'john@example.org', 2)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None, None, None, None)
(4, u'edyta1234', u'test', u'for234', None, None, 4, u'carl@nospam.net', 4)
(5, u'lepper123', u'lepp@qo.pl', u'0700LEP', None, None, None, None, None)
(6, u'Carla', u'clara@wp.pl', u'foobar', None, None, None, None, None)
>>>
# outer joiny są domyślnie "left outer join" czyli tabela po lewej + odpowiadające wartości z prawej tabeli.
# tutaj nie będzie rekordu z "Susane"
s = outerjoin(emails2, user_db).select()
run(s)
(1, u'mary@example.com', 1, 1, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None)
(2, u'john@nowhere.net', 2, 2, u'edyta', u'aw@wp.pl', u'foo123', None, None)
(3, u'john@example.org', 2, 2, u'edyta', u'aw@wp.pl', u'foo123', None, None)
(4, u'carl@nospam.net', 4, 4, u'edyta1234', u'test', u'for234', None, None)
>>>
Order By, Limit, Offset, distinct
>>> from sqlalchemy import *
>>> db=create_engine('oracle://django:password@xe')
>>> metadata=MetaData(db)
>>> metadata.echo=True
>>> def run(stmt):
... rs = stmt.execute()
... for row in rs:
... print row
...
>>> user_db=Table('user_sql',metadata,autoload=True)
#proste sortowanie
>>> s=user_db.select(order_by=[user_db.c.user_name])
>>> run(s)
(6, u'Carla', u'clara@wp.pl', u'foobar', None, None)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None)
(4, u'edyta1234', u'test', u'for234', None, None)
(5, u'lepper123', u'lepp@qo.pl', u'0700LEP', None, None)
(1, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None)
>>>
# rosnące, malejące sortowanie na wielu kolumnach
>>>s= user_db.select(user_db.c.user_name>'J', order_by=[desc(user_db.c.id), asc(user_db.c.user_name)])
>>>run(s)
(5, u'lepper123', u'lepp@qo.pl', u'0700LEP', None, None)
(4, u'edyta1234', u'test', u'for234', None, None)
(3, u'edyta123', u'ed@wp.pl', u'foo123', None, None)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None)
(1, u'zdzichu', u'sw@wp.pl', u'42zdzich', None, None)
>>>
# wybieranie niepowtarzalnych wierszy
>>>s = select([user_db.c.user_name], distinct=True)
>>>run(s)
(u'Carla',)
(u'edyta',)
(u'edyta123',)
(u'edyta1234',)
(u'lepper123',)
(u'zdzichu',)
>>>
# limit i offset
>>>s = user_db.select(offset=2, limit=2)
>>>run(s)
(6, u'Carla', u'clara@wp.pl', u'foobar', None, None)
(2, u'edyta', u'aw@wp.pl', u'foo123', None, None)
Update
# zmiana 'zdzichu' na 'krzychu'
>>> user_db.update(user_db.c.user_name=='zdzichu').execute(user_name='krzychu')
<sqlalchemy.engine.base.ResultProxy object at 0x02ACD650>
# przypisane wartości
>>>u = user_db.update(user_db.c.user_name==bindparam('name'),
values={'user_name':bindparam('newname')})
>>> u.execute(name='edyta', newname='ed')
<sqlalchemy.engine.base.ResultProxy object at 0x02B2C210>
# update jednej kolumny do drugiej
>>>user_db.update(values={user_db.c.password:user_db.c.user_name}).execute()
Delete
>>>user_db.delete(user_db.c.id==6).execute() <sqlalchemy.engine.base.ResultProxy object at 0x02ACD9D0> >>>