Created
November 4, 2016 17:08
-
-
Save green3g/435fd78be14c2f0459e078a3a14fa476 to your computer and use it in GitHub Desktop.
flask restless hybrid filtering
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
import flask | |
import flask_sqlalchemy | |
import flask_restless | |
from sqlalchemy.ext.hybrid import hybrid_property | |
from sqlalchemy import event, select, func | |
from sqlalchemy.sql.expression import case | |
# Create the Flask application and the Flask-SQLAlchemy object. | |
app = flask.Flask(__name__) | |
app.config['DEBUG'] = True | |
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db' | |
db = flask_sqlalchemy.SQLAlchemy(app) | |
class Workorder(db.Model): | |
""" | |
A workorder assigned to crew that may have one or many features | |
""" | |
__tablename__ = 'workorder' | |
id = db.Column(db.Integer, primary_key=True,) | |
features = db.relationship('Workorder_Feature', cascade='delete') | |
@hybrid_property | |
def is_complete(self): | |
""" | |
when all features are completed, this workorder should be completed | |
""" | |
features = Workorder_Feature.query.filter(Workorder_Feature.workorder_id == self.id) | |
for f in features: | |
if f.feature_status == 'Open': | |
return 'No' | |
return 'Yes' | |
@is_complete.expression | |
def is_complete(cls): | |
return case({True: 'Yes', False: 'No'}, | |
select([func.count(Workorder_Feature.id)]).\ | |
where(Workorder_Feature.workorder_id==cls.id).\ | |
where(Workorder_Feature.feature_status == 'Open') == 0 | |
) | |
class Workorder_Feature(db.Model): | |
""" | |
workorder features that are attached to a workorder. One workorder may have | |
many workorder features | |
""" | |
__tablename__ = 'workorder_feature' | |
id = db.Column(db.Integer, primary_key=True,) | |
workorder_id = db.Column( db.Integer(), db.ForeignKey('workorder.id')) | |
feature_status = db.Column(db.String(25)) | |
workorder = db.relationship('Workorder') | |
# Create the database tables. | |
db.drop_all() | |
db.create_all() | |
for a in range(5): | |
db.session.add(Workorder()) | |
db.session.commit() | |
for a in range(5): | |
db.session.add(Workorder_Feature(workorder_id=a, feature_status='Open')) | |
for a in range(5): | |
db.session.add(Workorder()) | |
db.session.commit() | |
for a in range(5): | |
db.session.add(Workorder_Feature(workorder_id=a, feature_status='Closed')) | |
db.session.commit() | |
# Create the Flask-Restless API manager. | |
manager = flask_restless.APIManager(app, flask_sqlalchemy_db=db) | |
# Create API endpoints, which will be available at /api/<tablename> by | |
# default. Allowed HTTP methods can be specified as well. | |
manager.create_api(Workorder, methods=['GET', 'POST', 'DELETE']) | |
manager.create_api(Workorder_Feature, methods=['GET']) | |
if __name__ == '__main__': | |
# start the flask loop | |
app.run(host='0.0.0.0') |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment