Python nie posiada określonej funkcji, aby sprawdzić, czy zmienna jest zdefiniowana, ponieważ jest założenie, że wszystkie zmienne zostały zdefiniowane przed użyciem. Próba uzyskania dostępu do zmiennej, która nie została wcześniej zdefiniowana będzie pluła wyjątkami.

W pythonie to rzadkość, żeby nie wiedzieć czy dana zmienna jest zdefiniowana. Jeżeli nastąpi już taka sytuacja, że tego nie wiesz możesz to sprawdzić poprzez próbę „try”, która w przypadku niezdefiniowanej zmiennej powinna wznieść wyjątek „NameError”:

try:
    test
except NameError:
    test = None

To nie wszystko, ponieważ musimy sprawdzić czy wyjątek został wzniesiony prostym warunkiem:

try:
    test
except NameError:
    test = None

if test is None:
    print "zmienna jest niezdefiniowana";
else:
    print "zmienna jest definiowana";

W wielu sytuacjach np. w formularzach wskazane jest korzystanie ze słowników i testów na obecność kluczy przy pomocy metody:

has_key()

Wstęp

SQLAlchemy jest zestawem pythonowych narzędzi SQL i ORM znacznie pomagającym programiście przy pracy z bazami danych.
Obsługuje następujące bazy danych: MySQL, SQLite, PostgreSQL, Oracle, MS-SQL i Firebird.


Instalacja

Jest kilka sposób instalacji, ze względu na prostotę polecam:

easy_install sqlalchemy

Jeżeli nie korzystasz z konta roota, prawdopodobnie będziesz musiał skorzystać z sudo.
Zastrzegam, że to nie jedyny sposób instalacji (;


ORM

ORM – ang. Object-Relational Mapping co w dosłownym tłumaczeniu znaczy Mapowanie Obiektowo-Relacyjne. Polega to mniej więcej na odwzorowaniu struktury obiektowej na relacyjnej bazie danych.


Pierwszy raz

Mniej więcej wyjaśniliśmy sobie kilka pojęć związanych z SQLAlchemy. Poznaliśmy prostą metodę instalacji i definicję ORM. Posiadając taką wiedzę czas na pokazanie i wytłumaczenie prostego przykładu „w kodzie”. Stwórz plik „przyklad.py„, zamieść w nim zawartość poniższego listingu i zapisz.

# -*- coding: utf-8 -*-
from sqlalchemy import *

db = create_engine('sqlite:///przyklad.db')

db.echo = False # Spróbuj zmienić na true (:. To taki odpowednik "verbose"

metadata = MetaData(db)

users = Table('users', metadata,
    Column('user_id', Integer, primary_key=True),
    Column('name', String(255)),
    Column('job', Integer),
    Column('pass', String(255)),
)
users.create() # Tworzymy tabelę

insert = users.insert()

insert.execute(name='Michal', job: 'Bezrobotny', pass='sekretne')

insert.execute({'name': 'Jan', 'job': 'Handlowiec', 'pass': 'tajne'},
          {'name': 'Kuba', 'job': 'Wozny'},
          {'name': 'Ola', 'job': 'Handlowiec'}) # Wkładamy dane

select = users.select()
result = select.execute()

wszystko = result.fetchall()
row = result.fetchone()
print 'Id:', row['user_id']
print 'Imie:', row['name']
print 'Praca:', row.job
print 'Haslo:', row['pass']

for row in wszystko: 
    print row.name, 'pracuje jako', row.job

Aby uruchomić nasz mały program wpisz w konsolę:

pyhton przyklad.py

Abyśmy mogli operwać na tabelach musieliśmy przypisać je do obiektu MetaData(). Teraz dopiero możemy stworzyć tabelę users. W przykładzie pokazałem dwa sposoby dodawania danych do bazy. Pierwszy pozwala na dodanie pojedyńczego obiektu, natomiast za pomocą drugiego sposobu podajemy listę słowników, żeby dodać wiele obiektów na raz. Jeżeli SQLAlchemy w naszych danych znajdzie się jakiś znak specjalny(np. apostrofę lub średnik) to SQLEngine automatycznie je zasleszuje. Dzięki takiej ochronie, atak SQL Injection staje się praktycznie niemożliwy. W SQLAlchemy pobieranie danych nie stanowi żadnego problemu, żeby pobrać jeden wiersz skorzystałem z funkcji fetchone(), natomiast żeby pobrać wszystkie rekordy na raz skorzystałem z fetchall() i wylistowałem je za pomocą pętli for, której obsługa w pythonie jest wyjątkowo prosta. Pewnie zwróciłeś uwagę to to, że korzystamy z sqlite. Jak już pewnie wiesz to nie jedyna baza obsługiwana przez SQLAlchemy. Poniżej przedstawiam sposoby łączenia się z innymi bazami danych:


db = create_engine('postgres://login:haslo@localhost:port/nazwabazydanych') # PostgreSQL
db = create_engine('mysql://login:haslo@localhost:port/nazwabazydanych') #MySQL
db = create_engine('oracle://login:haslo@localhost:port/nazwabazydanych') #Oracle

# Łączenie z bazami danych odbywa się za pomocą pewnego schematu:
# driver://login:haslo@host:port/nazwabazydanych

db = create_engine('sqlite:////bezwzgledna/sciezka/do/bazydanych.db') 
db = create_engine('sqlite:///relatywna/sciezka/do/bazydanych.db')
db = create_engine('sqlite://')  # baza w pamięci RAM



Pobieranie danych – szczegółowo

Poświęćmy trochę czasu, aby zbadać różne sposoby pobierania danych z bazy przy pomocy SQLAlchemy, które skracają czas pracy i pomagają programiście tworzyć zaawansowane zapytania z klauzulami WHERE, AND, OR itd. Na początek stwórzmy plik „pobieranie.py„.

# -*- coding: utf-8 -*-
from sqlalchemy import *

# Użyjmy tej samej bazy danych
db = create_engine('sqlite:///przyklad.db')

db.echo = True  # Robimy true, żeby widzieć co się dzieje

metadata = MetaData(db)

# Tabela users istnieje, ładujemy ją z bazy danych za pomocą "autoload"
users = Table('users', metadata, autoload=True)

def run(stmt):
    rs = stmt.execute()
    for row in rs:
        print row


# Zapytanie z  warunkiem jednego pola
q = users.select(users.c.name == 'Ola')
run(q)
q = users.select(users.c.id > 1 )
run(q)
q = users.select(users.c.job == 'Handlowiec'

# AND, OR i NOT
q = users.select(and_(users.c.id < 3, users.c.name != 'Jan'))
run(q)
q = users.select(or_(users.c.id > 1 , users.c.name != 'Jan'))
run(q)
q = users.select(not_(users.c.name == 'Jan'))
run(q)

# równie dobrze możemy korzystać z &, | i ~
q = users.select((users.c.job == 'Handlowiec') & (users.c.name != 'Ola'))
run(q)
q = users.select((users.c.job == 'Handlowiec' | (users.c.name != 'Kuba'))
run(q)
q = users.select(~(users.c.name == 'Michal'))
run(q)

# dostępne są również funkcje "like", "startswith", "endswith"
q = users.select(users.c.name.startswith('K'))
run(q)
q = users.select(users.c.name.like('%a%'))
run(q)
q = users.select(users.c.name.endswith('a'))
run(q)

# ponadto dostępne są "in" i "between"
q = users.select(users.c.id.between(1,3))
run(q)

# aby zapobiec konfliktowi z pythonem dajemy in_
q = users.select(users.c.name.in_('Jan', 'Ola'))
run(q)

# do wywoływanie funkcji SQL służy "func"
q = users.select(func.substr(users.c.name, 3, 1) == 'b')
run(q)

# pobieranie określonych 
q = select([users], users.c.job != 'Handlowiec')
run(q)
q = select([users.c.name, users.c.job], users.c.name != 'Michal')
run(q)

# korzystanie z funkcji SQL  "count"
q = select([func.count(users.c.user_id)])
run(q)

# korzystanie z funkcji SQL  "count", żeby zliczyć wszystko
q = select([func.count("*")], from_obj=[users])
run(q)

Order By, Limit i inne takie

Pobierając dane z bazy, zazwyczaj chcemy żeby były one odpowiednio posortowane. W takich przypadkach z pomocą przychodzą do nas Order by, Limit itp.

# -*- coding: utf-8 -*-
from sqlalchemy import *

# Użyjmy tej samej bazy danych
db = create_engine('sqlite:///przyklad.db')

db.echo = True  # Robimy true, żeby widzieć co się dzieje

metadata = MetaData(db)

# Tabela users istnieje, ładujemy ją z bazy danych za pomocą "autoload"
users = Table('users', metadata, autoload=True)

def run(stmt):
    rs = stmt.execute()
    for row in rs:
        print row

# proste wykorzystanie order_by
q = users.select(order_by=[users.c.name])
run(q)

# malejące sortowanie na podstawie id
q = users.select(order_by=[desc(users.c.user_id)])
run(q)

# rosnące sortowanie na podstawie id
q = users.select(order_by=[desc(users.c.user_id)])
run(q)

# sortowanie na wielu kolumnach
q = users.select(users.c.job>'A', order_by=[desc(users.c.user_id), asc(users.c.job)])

# pobieranie unikalnych danych
q = select([users.c.name], distinct=True)
run(q)

# wykorzystanie limit i offset
q = users.select(offset=2, limit=2)
run(q)

Delete

Czasami nie tylko chcemy pobierać dane, ale chcemy je usuwać. Tu z pomocą wkracza obiekt delete().

# -*- coding: utf-8 -*-
from sqlalchemy import *

# Użyjmy tej samej bazy danych
db = create_engine('sqlite:///przyklad.db')

db.echo = True  # Robimy true, żeby widzieć co się dzieje

metadata = MetaData(db)

# Tabela users istnieje, ładujemy ją z bazy danych za pomocą "autoload"
users = Table('users', metadata, autoload=True)

def run(stmt):
    stmt.execute()

# Usuwamy rekord o id równym 1
q = users.delete(users.c.user_id == 1)
run(q)

# Usuwamy rekordy gdzie zawodem jest "Handlowiec"
q = users.delete(users.c.job == 'Handlowiec')
run(q)

# Usuwanie rekordów dla wielu argumentów
users.delete().execute(
		{'name': '%Jan%'},
		{'name': '%Ola%'},
		{'job': '%wozny%'}
	)

	

Update

Oprócz usuwania często chcemy coś poprawić lub zamienić w naszych rekordach. Pomoże nam w tym obiekt update().

# -*- coding: utf-8 -*-
from sqlalchemy import *

# Użyjmy tej samej bazy danych
db = create_engine('sqlite:///przyklad.db')

db.echo = True  # Robimy true, żeby widzieć co się dzieje

metadata = MetaData(db)

# Tabela users istnieje, ładujemy ją z bazy danych za pomocą "autoload"
users = Table('users', metadata, autoload=True)

# Podmieniamy "Kuba" na "Marcin"
users.update(users.c.user_name=='Kuba').execute(user_name='Marcin')

# Update dla wielu argumentów
users.update().execute(
        {'user_id':1, 'user_name':'Helena', 'job':'wozna'},
        {'user_id':2, 'user_name':'Zdzislaw', 'job':'ksiegowy'},
        {'user_id':3, 'user_name':'Maciek', 'job':'sportowiec'},
    )

JOIN czyli łączenie tabel

Wielokrotnie spotykamy się z problemem struktury bazy danych. Np. Trzymania w innej tabeli adresów zamieszkania naszych „userów”. Na początek stwórzmy sobie nowy plik „jointest.py” i wypełnij go poniższą zawartością:

# -*- coding: utf-8 -*-
from sqlalchemy import *

# Tworzymy nową bazę danych
db = create_engine('sqlite:///przykladjoin.db')

db.echo = True  # Robimy true, żeby widzieć co się dzieje

metadata = MetaData(db)

users = Table('users', metadata,
    Column('user_id', Integer, primary_key=True),
    Column('name', String(255)),
    Column('job', Integer),
    Column('pass', String(255)),
)

users.create() # Tworzymy tabelę users

address = Table('addresses', metadata,
	Column('address_id', Integer, primary_key=True),
	Column('town', String),
	Column('user_id', Integer, ForeignKey('users.user_id')),
	)

address.create() # Tworzymy tabelę address

insert = users.insert()
insert.execute(
			{'name': 'Jan', 'job': 'Handlowiec', 'pass': 'tajne'},
			{'name': 'Kuba', 'job': 'Wozny', 'pass': 'tajne'},
			{'name': 'Ola', 'job': 'Handlowiec', 'pass': 'tajne'},
			{'name': 'Karolina', 'job': 'Sportowiec', 'pass': 'tajne'},
			{'name': 'Marcin', 'job': 'Wozny', 'pass': 'tajne'}
		) # Wkładamy dane

insert = address.insert()
insert.execute(	
			{'town': 'Warsaw', 'user_id': 1},
			{'town': 'New York', 'user_id': 2},
			{'town': 'London', 'user_id': 3},
			{'town': 'Oslo', 'user_id': 4}
		) # Wkładamy dane


def run(stmt):
    rs = stmt.execute()
    for row in rs:
        print row	
	
# Spróbujmy trochę nabroić i zróbmy pełny join (:
# ze względu na brak klauzuli WHERE
# program zwróci nam wszystkie możliwe kombinacje rekordów.
q = select([users, address])	
run(q)

# Teraz spróbujemy z klauzulą WHERE 
# SQL: bla bla WHERE addresses.user_id = users.user_id
q = select([users, address], address.c.user_id == users.c.user_id)
run(q)

# Pobieramy tylko kilka kolumn
q = select([users.c.name, address.c.town], address.c.user_id == users.c.user_id)
run(q)

# sqlalchemy oferuje "mądry" obiekt "join"
# który określa odpowiednie parametry łączenia
# na podstawie "foreign keys" tabel
q = join(users, address).select()
run(q)

# jeżeli chcemy pobrać wszystkich userów
# nawet jeżeli nie mają wpisanego miasta
# to korzystamy z outerjoin w takiej kolejności 
q = outerjoin(users, address).select()
run(q)

# kolejność ma znaczenie
# domyślnie outerjoin jest "left outer join" 
# to znaczy tabela po lewej i odpowiadające jej wartości z tabeli po prawej
# w ten sposób nie pobierzemy rekordu z "Marcinem"
# ponieważ nie ma jego id w tabeli addresses
q = outerjoin(address, users).select()
run(q)

i wykonaj takie polecenie:

python jointest.py

Obiekt bindparam()

W SQLAlchemy możesz sam „zbindować„(określić) inną nazwę dla istniejącej nazwy jakiejś kolumny i korzystać z takiego „binda” wielokrotnie w swoich zapytaniach do bazy. Brzmi to strasznie, ale wytłumaczenie tego słowami ( przynajmniej dla mnie ) jest stosunkowo trudne, dlatego zapraszam do zajrzenia do kodu:

q = users.select(users.c.name==bindparam('username'))
conn.execute(q, username = 'Jan').fetchall() 

q = users.select(users.c.user_id==bindparam('id'))
conn.execute(q, id = 1 ).fetchall() 

Mam nadzieję, że jakoś to pojęliście (: Dokumentacja


Union

Union służy do łączenia wyników selectów ( nie pokaże powtarzających się wierszy ):

# -*- coding: utf-8 -*-
from sqlalchemy import *

# Użyjmy tej samej bazy danych
db = create_engine('sqlite:///przykladjoin.db')

db.echo = True  # Robimy true, żeby widzieć co się dzieje

metadata = MetaData(db)

# Tabela users istnieje, ładujemy ją z bazy danych za pomocą "autoload"
users = Table('users', metadata, autoload=True)
address = Table('addresses', metadata, autoload=True)

union(
    address.select(address.c.street=='New York'),
    address.select(address.c.street=='Oslo'),
    address.select(address.c.street=='Warsaw'),
    order_by=[addresses.c.street]
).execute()

users.select(
    users.c.user_id==1
  ).union_all(
      users.select(
          users.c.user_id==3
      ), 
      order_by=[users.c.user_id]
  ).execute()

Podsumowanie

SQLAlchemy to narzędzie proste, szybkie i wygodne, przyspieszające pracę programisty, który nie musi się już zamartwiać długimi zapytaniami z JOIN itp. Zdecydowanie POLECAM!( zwłaszcza do pracy z Pylonsem )


Źródła wiedzy