PythonTracks

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:

• Postgres: psycopg2
• SQLite: pysqlite
• MySQL: MySQLDB
• Oracle: cx_Oracle

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>
>>>

Przydatne Linki

• SQLAlchemy 0.7 Documentation

• Essential SQLAlchemy

• Using SQLAlchemy with Django

• ORM