Created
May 7, 2011 19:04
-
-
Save k4ml/960743 to your computer and use it in GitHub Desktop.
Flask+SQLAlchemy
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment