SQLAlchemy ORM及Core操作方法

SQLAlchemy ORM 详解

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker

print(sqlalchemy.__version__)
# # examples of connection http://docs.sqlalchemy.org/en/rel_1_0/core/engines.html#sqlalchemy.create_engine
engine = create_engine(‘sqlite:///foo.db’, echo=True)
#echo:是否显示原始sql语句

# Base = declarative_base()
#
# class User(Base):
#     __tablename__ = ‘users’
#
#     id = Column(Integer, primary_key=True)
#     name = Column(String)
#     fullname = Column(String)
#     password = Column(String)
#
#     def __repr__(self):
#        return “<User(name=’%s’, fullname=’%s’, password=’%s’)>” % (
#                             self.name, self.fullname, self.password)
#
# Base.metadata.create_all(engine)  #
创建表

# ed_user = User(name=’ed’, fullname=’Ed Jones’, password=’edspassword’)
# print(ed_user)
#

##事务
# Session = sessionmaker(bind=engine)
# session = Session()
# session.add(ed_user)
# our_user = session.query(User).filter_by(name=’ed’).first()
# SELECT * FROM users WHERE name=”ed” LIMIT 1;

# session.add_all([
#     User(name=’wendy’, fullname=’Wendy Williams’, password=’foobar’),
#     User(name=’mary’, fullname=’Mary Contrary’, password=’xxg527′),
#     User(name=’fred’, fullname=’Fred Flinstone’, password=’blah’)])
#
# session.commit()

# print(session.query(User).filter_by(name=’ed’).first())
# print(session.query(User).all())
# for row in session.query(User).order_by(User.id):
#     print(row)
# for row in session.query(User).filter(User.name.in_([‘ed’, ‘wendy’, ‘jack’])):
#     print(row)
# for row in session.query(User).filter(~User.name.in_([‘ed’, ‘wendy’, ‘jack’])):
#     print(row)
# print(session.query(User).filter(User.name == ‘ed’).count())
#
# from sqlalchemy import and_, or_
# for row in session.query(User).filter(and_(User.name == ‘ed’, User.fullname == ‘Ed Jones’)):
#     print(row)
# for row in session.query(User).filter(or_(User.name == ‘ed’, User.name == ‘wendy’)):
#     print(row)
#
# from sqlalchemy import ForeignKey
# from sqlalchemy.orm import relationship, backref
#
# class Address(Base):
#     __tablename__ = ‘addresses’
#     id = Column(Integer, primary_key=True)
#     email_address = Column(String, nullable=False)
#     user_id = Column(Integer, ForeignKey(‘users.id’))
#
#     user = relationship(“User”, backref=backref(‘addresses’, order_by=id))
#
#     def __repr__(self):
#         return “<Address(email_address=’%s’)>” % self.email_address
# # Base.metadata.create_all(engine)
# #
# jack = User(name=’jack’, fullname=’Jack Bean’, password=’gjffdd’)
# jack.addresses = [
#                 Address(email_address=’jack@google.com’),
#                 Address(email_address=’j25@yahoo.com’)]
# session.add(jack)
# session.commit()
#
# for u, a in session.query(User, Address).\
#                     filter(User.id==Address.user_id).\
#                     filter(Address.email_address==’jack@google.com’).\
#                     all():
#     print u, a

SQLAlchemy Core 详解

import sqlalchemy
print(sqlalchemy.__version__)

from sqlalchemy import create_engine
from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey

engine = create_engine(‘sqlite:///foo.db’, echo=True)

metadata = MetaData()
users = Table(‘users’, metadata,
    Column(‘id’, Integer, primary_key=True),
    Column(‘name’, String),
    Column(‘fullname’, String),
)

addresses = Table(‘addresses’, metadata,
  Column(‘id’, Integer, primary_key=True),
  Column(‘user_id’, None, ForeignKey(‘users.id’)),
  Column(’email_address’, String, nullable=False)
)

# metadata.create_all(engine)
conn = engine.connect()

# conn.execute(users.insert(), [dict(name=’jack’, fullname=’Jack Jones’),
#                               dict(name=’wendy’, fullname=’Wendy Williams’)])
# conn.execute(addresses.insert(), [
#    {‘user_id’: 1, ’email_address’ : ‘jack@yahoo.com’},
#    {‘user_id’: 1, ’email_address’ : ‘jack@msn.com’},
#    {‘user_id’: 2, ’email_address’ : ‘www@www.org’},
#    {‘user_id’: 2, ’email_address’ : ‘wendy@aol.com’},
# ])

from sqlalchemy.sql import select
# s = select([users])
# result = conn.execute(s)
# for row in result:
#     print(row)

# s = select([users, addresses]).where(users.c.id == addresses.c.user_id)
# for row in conn.execute(s):
#     print row

from sqlalchemy.sql import text
s = text(
    “SELECT users.fullname || ‘, ‘ || addresses.email_address AS title ”
        “FROM users, addresses ”
        “WHERE users.id = addresses.user_id ”
        “AND users.name BETWEEN 😡 AND :y ”
        “AND (addresses.email_address LIKE :e1 ”
            “OR addresses.email_address LIKE :e2)”
)
print(conn.execute(s, x=‘m’, y=‘z’, e1=‘%@aol.com’, e2=‘%@msn.com’).fetchall())