Usando sqlalchemy
Usando somente a query expression language
db.py
1 from sqlalchemy import create_engine, Table, Column, MetaData
2 from sqlalchemy import Boolean, Integer, Unicode, String
3 from sqlalchemy.engine.url import URL
4
5 ### para conexao no sqlite:
6 _url = 'sqlite://'
7 ### para conexao no firebird
8 # _url = URL('firebird', 'SYSDBA', 'masterkey', '192.168.1.11', '3052', 'bdband')
9 ### para conexao no mysql
10 # _url = 'mysql://usuario:senha@servidor/banco'
11
12 # cria o engine e metadata
13 engine = create_engine(_url)
14 metadata = MetaData(bind=engine)
15
16 #cria as tabelas
17 tb_usuarios = Table('usuarios', metadata,
18 Column('id', Integer, primary_key=True),
19 Column('nome', Unicode(20), nullable=False),
20 Column('senha', String(10), nullable=False),
21 Column('ativo', Boolean, nullable=False)
22 )
23
24 #cria as tabelas no banco (caso nao existam)
25 metadata.create_all()
Exemplo de uso do arquivo acima
Um outro exemplo de uso, desta vez para consulta, e usando o contextlib.closing para fechar a conexão:
1 import db
2 from contextlib import closing
3
4 with closing(db.engine.connect()) as con:
5 sel_usuarios = db.tb_usuarios.select(db.tb_usuarios.c.ativo)
6 resultado = con.execute(sel_usuarios)
7 for registro in resultado:
8 print registro.nome, registro.id, registro.senha, registro.ativo
9 # conexao fechada automaticamente, mesmo se der algum erro
Usando o ORM e classes mapeadas
db.py
1 from sqlalchemy import create_engine, Table, Column, MetaData
2 from sqlalchemy import Boolean, Integer, Unicode, String
3 from sqlalchemy.engine.url import URL
4 from sqlalchemy.orm import sessionmaker, mapper
5 import random
6 import string
7
8 ### para conexao no sqlite:
9 _url = 'sqlite://'
10 ### para conexao no firebird
11 # _url = URL('firebird', 'SYSDBA', 'masterkey', '192.168.1.11', '3052', 'bdband')
12 ### para conexao no mysql
13 # _url = 'mysql://usuario:senha@servidor/banco'
14
15 # cria o engine e metadata
16 engine = create_engine(_url, echo=True)
17 metadata = MetaData(bind=engine)
18
19 #cria as tabelas
20 tb_usuarios = Table('usuarios', metadata,
21 Column('id', Integer, primary_key=True),
22 Column('nome', Unicode(20), nullable=False),
23 Column('senha', String(10), nullable=False),
24 Column('ativo', Boolean, nullable=False)
25 )
26
27 #cria as classes
28 class Usuario(object):
29 def __init__(self, nome, senha=None, ativo=False):
30 if senha is None:
31 # gera uma senha aleatoria
32 senha = ''.join(random.choice(string.letters + string.digits)
33 for x in xrange(10))
34 self.nome = nome
35 self.senha = senha
36 self.ativo = ativo
37
38 # um metodo simples para demontrar que a classe pode ter quaisquer metodos
39 def verifica_senha(self, senha_verificar):
40 return self.senha == senha_verificar
41
42 # mapeia a classe -> tabela
43 mapper(Usuario, tb_usuarios)
44
45 #cria as tabelas no banco (caso nao existam)
46 metadata.create_all()
47
48 #cria o sessionmaker
49 Session = sessionmaker(bind=engine)
Usando:
1 import db
2
3 #posso criar uma instancia da minha classe
4 u = db.Usuario(u'big_rid')
5 # alterar atributos
6 u.senha = '123456'
7
8 # na hora de usar pega uma sessao
9 s = db.Session()
10 s.add(u) # adiciona o objeto a sessao
11 s.add_all([ # para adicionar varios objetos mesmo tempo
12 db.Usuario(nome=u'jack', ativo=True),
13 db.Usuario(nome=u'jonh', senha='senha'),
14 ])
15 s.commit()
16
17 # Posso alterar minha instancia, o sqlalchemy gera um UPDATE automaticamente
18 u.ativo = True
19 s.commit() # UPDATE usuarios SET ativo = 1 WHERE id = ?
20
21 s.close()
Consultando (queries)
1 from contextlib import closing
2 import db
3
4 with closing(db.Session()) as s:
5 # uma query retornara instancias da classe definida acima
6 for usu in s.query(db.Usuario).filter(db.Usuario.ativo == True):
7 print usu.nome, usu.id, usu.senha, usu.ativo
8 # chamo um metodo customizado no objeto retornado:
9 if usu.verifica_senha('123456'):
10 # posso alterar o objeto, sqlalchemy gera UPDATE
11 usu.senha = '321321'
12 print 'senha alterada'
13 s.commit()
14
15 # sessao fechada automaticamente, mesmo se der erro. (no caso de erro
16 # um ROLLBACK sera dado na transacao automaticamente)
Usando declarative para definir a classe e a tabela ao mesmo tempo
db.py
1 from sqlalchemy import create_engine, Table, Column, MetaData
2 from sqlalchemy import Boolean, Integer, Unicode, String
3 from sqlalchemy.engine.url import URL
4 from sqlalchemy.orm import sessionmaker, mapper
5 from sqlalchemy.ext.declarative import declarative_base
6 import random
7 import string
8
9 ### para conexao no sqlite:
10 _url = 'sqlite://'
11 ### para conexao no firebird
12 # _url = URL('firebird', 'SYSDBA', 'masterkey', '192.168.1.11', '3052', 'bdband')
13 ### para conexao no mysql
14 # _url = 'mysql://usuario:senha@servidor/banco'
15
16 # cria o engine e o declarative_base
17 engine = create_engine(_url, echo=True)
18
19 Base = declarative_base(bind=engine)
20
21
22 #cria as classes e tabelas ja mapeando
23 class Usuario(Base):
24 __tablename__ = 'usuarios'
25
26 id = Column(Integer, primary_key=True)
27 nome = Column(Unicode(20), nullable=False)
28 senha = Column(String(10), nullable=False)
29 ativo = Column(Boolean, nullable=False)
30
31 def __init__(self, nome, senha=None, ativo=False):
32 if senha is None:
33 # gera uma senha aleatoria
34 senha = ''.join(random.choice(string.letters + string.digits)
35 for x in xrange(10))
36 self.nome = nome
37 self.senha = senha
38 self.ativo = ativo
39
40 # um metodo simples para demontrar que a classe pode ter quaisquer metodos
41 def verifica_senha(self, senha_verificar):
42 return self.senha == senha_verificar
43
44 #cria as tabelas no banco (caso nao existam)
45 Base.metadata.create_all()
46
47 #cria o sessionmaker
48 Session = sessionmaker(bind=engine)
A utilização pode ser exatamente como no caso anterior - a diferença é somente na declaração, que desta forma é mais fácil e concisa.