01_engine_usage

from sqlalchemy import create_engine
import os

if os.path.exists("some.db"):
    os.remove("some.db")
engine = create_engine("sqlite:///some.db")
engine.execute("""
    create table employee (
        emp_id integer primary key,
        emp_name varchar
    )
""")
engine.execute("insert into employee (emp_name) values (:emp_name)", emp_name="dilbert")

result = engine.execute("select * from employee")
print(result.fetchall())

-----

02_metadata

from sqlalchemy import MetaData
from sqlalchemy import Table, Column
from sqlalchemy import Integer, String
from sqlalchemy import String, Numeric, DateTime, Enum,Unicode,UnicodeText
from sqlalchemy import ForeignKey,ForeignKeyConstraint,create_engine
import os
from sqlalchemy import inspect
e = create_engine("sqlite:///some.db")
metaData=MetaData()
user_table = Table('user', metaData,
                   Column('id', Integer, primary_key=True),
                   Column('name', String),
                   Column('fullname', String)
                   )
addresses_table = Table('address', metaData,
                        Column('id', Integer, primary_key=True),
                        Column('email_address', String(100), nullable=False),
                        Column('user_id', Integer, ForeignKey('user.id'))
                        )
story_table = Table('story', metaData,
                    Column('story_id', Integer, primary_key=True),
                    Column('version_id', Integer, primary_key=True),
                    Column('headline', Unicode(100), nullable=False),
                    Column('body', UnicodeText)
                    )
published_table = Table('published', metaData,
                        Column('pub_id', Integer, primary_key=True),
                        Column('pub_timestamp', DateTime, nullable=False),
                        Column('story_id', Integer),
                        Column('version_id', Integer),
                        ForeignKeyConstraint(
                            ['story_id', 'version_id'],
                            ['story.story_id', 'story.version_id'])
                        )

network_table= Table('network',metaData,
             Column('network_id', Integer, primary_key=True),
             Column('name', String(100), nullable=False),
             Column('created_at', DateTime, nullable=False),
             Column('owner_id', Integer,ForeignKey('user.id'))
             )
metaData.create_all(e)
metadata2=MetaData()
reflect = Table('user', metadata2, autoload=True, autoload_with=e)
inspector = inspect(e)
columns=inspector.get_columns('network')
for column in columns:
    print(column)
results = []
table_names=inspector.get_table_names()
for table_name in table_names:
   for column in inspector.get_columns(table_name):
       if column['name'] == 'story_id':
           results+=table_name
print(results)

Result:

{'primary_key': 1, 'nullable': False, 'default': None, 'name': 'network_id', 'type': INTEGER(), 'autoincrement': True}
{'primary_key': 0, 'nullable': False, 'default': None, 'name': 'name', 'type': VARCHAR(length=100), 'autoincrement': True}
{'primary_key': 0, 'nullable': False, 'default': None, 'name': 'created_at', 'type': DATETIME(), 'autoincrement': True}
{'primary_key': 0, 'nullable': True, 'default': None, 'name': 'owner_id', 'type': INTEGER(), 'autoincrement': True}
published
story

-----

03_sql_expressions

metadata = MetaData()
engine = create_engine("sqlite:///some.db")

user_table = Table('user', metadata,
                    Column('id', Integer, primary_key=True),
                    Column('username', String(50)),
                    Column('fullname', String(50))
                   )
metadata.create_all(engine)

print(user_table.c.fullname == 'ed')
print(and_(user_table.c.fullname == 'ed',user_table.c.id > 5))
print(or_(user_table.c.username == 'ed'))

conn = engine.connect()
result = conn.execute(user_table.insert(), [{'username': 'dilbert', 'fullname': 'Dilbert Jones'}])
print(result.inserted_primary_key)

select_some_values = select([user_table]).\
                    where(
                        or_(
                            user_table.c.username == 'wendy',
                            user_table.c.username == 'dilbert'
                        )
                    ).\
                    order_by(user_table.c.fullname)
conn.execute(select_some_values).fetchall()

addressTable = Table("address", metadata,
                      Column('id', Integer, primary_key=True),
                      Column('user_id', Integer, ForeignKey('user.id'),
                             nullable=False),
                      Column('email_address', String(100), nullable=False)
                      )
metadata.create_all(engine)
conn.execute(addressTable.insert(), [
    {"user_id": 1, "email_address": "ed@ed.com"},
    {"user_id": 1, "email_address": "ed@gmail.com"},
    {"user_id": 2, "email_address": "jack@yahoo.com"},
    {"user_id": 3, "email_address": "wendy@gmail.com"},
])

select_query = select([user_table.c.fullname, address_table.c.email_address]).\
    select_from(user_table.join(address_table)).\
    where(user_table.c.username == 'ed').\
    order_by(address_table.c.email_addres)
print(select_query)

result = user_table.update().values(fullname="Ed Jones").where(user_table.c.username == 'ed')
conn.execute(result)
print(result.rowcount)

bonus_query = user_table.update()
                        .values(fullname = user_table.c.fullname + user_email.as_scalar())
                        .where(user_table.c.username
                        .in_({'jack', 'wendy'}))
conn.execute(bonus_query)

-----

04_orm

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String,ForeignKey,Numeric
from sqlalchemy.orm import Session,relationship
from sqlalchemy import create_engine

engine = create_engine("sqlite:///some.db")
conn = engine.connect()

Base = declarative_base()

class Network(Base):
    __tablename__ = 'network'
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    def __repr__(self):
        return "Network: %r" % (self.name)
Base.metadata.create_all(engine)

from sqlalchemy.orm import Session
session = Session(bind = engine)
session.add(Network(name = 'net1'))
session.add(Network(name = 'net2'))
session.commit()
select_network = select([Network.__table__])
result = conn.execute(select_network)
print (result.fetchall())

class User(Base):
    __tablename__ = 'user'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    def __repr__(self):
        return "<User(%r, %r)>" % (
                self.name, self.fullname
            )

session.add_all([
    User(name='wendy', fullname='Wendy Weathersmith'),
    User(name='mary', fullname='Mary Contrary'),
    User(name='fred', fullname='Fred Flinstone')
])

query = session.query(User.fullname).order_by(User.fullname)
print(query.all())

query2 = query.filter(User.name.in_(['mary', 'ed']))
query2.all()
print(query2[1])

query3 = session.query(User.name, Address.email_address).join(Address).filter(Address.email_address.in_(['j25@yahoo.com']))
print(query3.all())

class Account(Base):
    __tablename__ = 'account'
    id = Column(Integer, primary_key=True)
    owner = Column(String(50), nullable=False)
    balance = Column(Numeric, default=0)
    def __repr__(self):
        return "Account: %r, %r" % (self.owner, self.balance)

class Transaction(Base):
    __tablename__ = 'transaction'
    id = Column(Integer, primary_key=True)
    amount = Column(Numeric, nullable=False)
    account_id = Column(Integer, ForeignKey(Account.__tablename__ + '.id'), nullable=False)
    account = relationship('Account', backref="transactions")
    def __repr__(self):
        return "Transaction: %r" % (self.amount)

session = Session(bind = engine)
session.add_all([Account(owner = "Jack Jones", balance = 5000),
                 Account(owner="Ed Rendell", balance=10000),
                 Transaction(amount=500, account=account1),
                 Transaction(amount=4500, account=account1),
                 Transaction(amount=6000, account=account2),
                 Transaction(amount = 4000,  account = account2)])
session.commit()

for account in session.query(Account).all():
    owner = account.owner
    balance = account.balance
    spent_money = 0
    for account_transaction in account.transactions:
        spent_money += account_transaction.amount

    print("Account owner: " + str(owner) + '\t' +
          "Account balance: " + str(balance) + '\t' +
          "Spent money: " + str(spent_money))

Result:

Account owner: Jack Jones   Account balance: 5000.0000000000    Spent money: 5000.0000000000
Account owner: Ed Rendell   Account balance: 10000.0000000000   Spent money: 10000.0000000000