import datetime
from sqlalchemy.engine import create_engine
from sqlalchemy.ext.declarative.api import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.orm.session import Session
from sqlalchemy.schema import Column, ForeignKey
from sqlalchemy.sql.expression import case
from sqlalchemy.sql.functions import count
from sqlalchemy.types import Integer, String, DateTime
Base = declarative_base()
class Book(Base):
__tablename__ = 'book'
id = Column(Integer, primary_key=True)
title = Column(String)
publication_date = Column(DateTime)
pages_count = Column(Integer)
author_id = Column(Integer, ForeignKey('author.id'))
class Author(Base):
__tablename__ = 'author'
id = Column(Integer, primary_key=True)
name = Column(String)
books = relationship(Book, backref="author")
db_uri = "sqlite:///"
engine = create_engine(db_uri, echo=False)
Base.metadata.create_all(engine)
session = Session(bind=engine)
insert_stmt = Book.__table__.insert()
print insert_stmt
INSERT INTO book (id, title, publication_date, pages_count, author_id) VALUES (:id, :title, :publication_date, :pages_count, :author_id)
data = {'title': 'Essential SQLAlchemy', 'publication_date': datetime.datetime.strptime('2010-05-05', '%Y-%m-%d'),
'pages_count': 240}
engine.connect().execute(insert_stmt, data)
<sqlalchemy.engine.result.ResultProxy at 0x4c87f90>
session = Session(bind=engine)
q = session.query(Book).filter(Book.title == 'Essential SQLAlchemy')
print q
SELECT book.id AS book_id, book.title AS book_title, book.publication_date AS book_publication_date, book.pages_count AS book_pages_count, book.author_id AS book_author_id FROM book WHERE book.title = :title_1
book = q.one()
print (book.id, book.title)
(1, u'Essential SQLAlchemy')
book.__dict__
{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState at 0x4c87570>, 'author_id': None, 'id': 1, 'pages_count': 240, 'publication_date': datetime.datetime(2010, 5, 5, 0, 0), 'title': u'Essential SQLAlchemy'}
author = Author(name='Rick Copeland')
author.books.append(book)
print author.books
[<__main__.Book object at 0x04C7EAB0>]
session.add(book)
session.flush()
is_novel_column = case([(Book.pages_count>200, 1)], else_=0)
print is_novel_column
CASE WHEN (book.pages_count > :pages_count_1) THEN :param_1 ELSE :param_2 END
novel_query = session.query(is_novel_column.label('is_alias'), count()).\
group_by(is_novel_column).\
order_by(is_novel_column)
print novel_query
SELECT CASE WHEN (book.pages_count > :pages_count_1) THEN :param_1 ELSE :param_2 END AS is_alias, count(*) AS count_1 FROM book GROUP BY CASE WHEN (book.pages_count > :pages_count_1) THEN :param_1 ELSE :param_2 END ORDER BY CASE WHEN (book.pages_count > :pages_count_1) THEN :param_1 ELSE :param_2 END
print novel_query.all()
[(1, 1)]
session.close()