test_flask.py

from flask import Flask, request, session
from sqlalchemy import create_engine, Table, MetaData
from sqlalchemy.orm import mapper, relationship, sessionmaker, scoped_session
from sqlalchemy.sql import select

app = Flask(__name__)

engine = create_engine('postgresql+psycopg2://kamal@localhost/clinic_dev')
metadata = MetaData(bind=engine)
DBSession = scoped_session(sessionmaker(bind=engine))

customers_table = Table('customers', metadata, autoload=True, schema='cvt')
orders_table = Table('orders', metadata, autoload=True, schema='cvt')
order_items_table = Table('order_items', metadata, autoload=True, schema='cvt')
items_table = Table('items', metadata, autoload=True, schema='cvt')
sessions_table = Table('sessions', metadata, autoload=True)
users_table = Table('users', metadata, autoload=True)

class Model(object):
    query = DBSession.query_property()

class Customer(Model):
    pass

class Order(Model):
    pass

class Item(Model):
    pass

class OrderItem(Model):
    def __str__(self):
        return self.item.name

class User(Model):
    pass

mapper(Customer, customers_table, properties={
    'orders': relationship(Order)
})
mapper(Order, orders_table, properties={
    'items': relationship(OrderItem),
    'customer': relationship(Customer, uselist=False),
})
mapper(OrderItem, order_items_table, properties={
    'orders': relationship(Order),
    'item': relationship(Item, uselist=False),
})
mapper(Item, items_table)
mapper(User, users_table)

DRUPAL_SESSION_NAME = 'SESS6e9eabbb344f405db25d0412846cef39'

def is_login():
    sid = request.cookies.get(DRUPAL_SESSION_NAME, '')
    if not sid:
        return False
    s = select([sessions_table], sessions_table.c.sid==sid).execute()
    row = s.fetchone()
    if not row:
        return False
    if row['uid'] > 0:
        session['drupal_uid'] = row['uid']
        return True
    return False

def current_user():
    try:
        uid = session['drupal_uid']
    except KeyError:
        uid = None

    if uid is None:
        return False

    user = User.query.filter(User.uid==uid).one()
    if user:
        return user
    return False

@app.route('/')
def hello():
    out = []
    for customer in dbsession.query(Customer).filter(Customer.name.like('%Yusoff%')):
        out.append("%s %s" % (customer.id, customer.name))
        for order in customer.orders:
            out.append("\t #%d %s" % (order.id, order.transaction_date.strftime('%d-%m-%Y')))
            for item in order.items:
                out.append("\t\t %s RM%.2f" % (item.item.name, item.price))
    return "".join(out)

@app.route('/cookies')
def cookies():
    if not is_login():
        return 'Not login'
    user = current_user()
    out = []
    out.append("Login as %s" % user.name)
    for key, value in request.cookies.items():
        out.append("%s=%s" % (key, value))
    return "<br />".join(out)

@app.route('/add/order/')
def add_order():
    from formalchemy import FieldSet, Grid
    order1 = Order.query.filter(Order.id==202).one()
    fs = FieldSet(order1)
    fs.configure(exclude=[fs.customer, fs.items])
    return fs.render()

@app.after_request
def shutdown_session(response):
    DBSession.remove()
    return response

app.debug = True
app.secret_key = 'secret'
if __name__ == '__main__':
    app.run()