-
-
Save hondajojo/afa7b33cbe629fc18877e0be90eb29ef to your computer and use it in GitHub Desktop.
flask restful api with flask_sqlalchemy
This file contains hidden or 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
#!/usr/bin/env python | |
# -*- coding:utf-8 -*- | |
# Created on 2016-09-21 11:21:24 | |
from flask import Flask, jsonify, abort, make_response | |
from flask import request | |
from flask_sqlalchemy import SQLAlchemy | |
import json | |
app = Flask(__name__) | |
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:1@localhost/test' | |
db = SQLAlchemy(app) | |
class Articles(db.Model): | |
__tablename__ = 'articles' | |
id = db.Column(db.Integer, primary_key=True) | |
title = db.Column(db.String(100), unique=True) | |
content = db.Column(db.Text) | |
def __init__(self, title, content): | |
self.title = title | |
self.content = content | |
def __repr__(self): | |
return '<Articles %r>' % self.title | |
def as_dict(self): | |
return {c.name: getattr(self, c.name) for c in self.__table__.columns} | |
@app.route('/blog/api/articles', methods=['GET', 'POST']) | |
def get_articles(): | |
if request.method == "GET": | |
articles = json.dumps([article.as_dict() | |
for article in Articles.query.all()]) | |
return jsonify({'articles': articles}) | |
elif request.method == "POST": | |
if not request.json or not 'title' in request.json: | |
abort(404) | |
try: | |
article = Articles(title=request.json[ | |
'title'], content=request.json.get('content', '')) | |
db.session.add(article) | |
db.session.commit() | |
articles = json.dumps([article.as_dict() | |
for article in Articles.query.all()]) | |
return jsonify({'articles': articles}), 201 | |
except Exception as e: | |
return jsonify({'result': False, 'message': str(e)}) | |
@app.route('/blog/api/articles/<int:article_id>', methods=['GET']) | |
def get_article(article_id): | |
article = Articles.query.filter_by(id=article_id).first_or_404().as_dict() | |
return jsonify({'article': article}) | |
@app.route('/blog/api/articles/<int:article_id>', methods=['PUT']) | |
def update_article(article_id): | |
article = Articles.query.filter_by(id=article_id).first_or_404() | |
if not request.json: | |
abort(400) | |
article.title = request.json.get('title', article.title) | |
article.content = request.json.get('content', article.content) | |
db.session.add(article) | |
db.session.commit() | |
return jsonify({'article': article.as_dict()}) | |
@app.route('/blog/api/articles/<int:article_id>', methods=['DELETE']) | |
def delete_article(article_id): | |
article = Articles.query.filter_by(id=article_id).first_or_404() | |
db.session.delete(article) | |
db.session.commit() | |
return jsonify({'result': True}) | |
@app.errorhandler(404) | |
def not_found(error): | |
return make_response(jsonify({'error': 'Not found'}), 404) | |
if __name__ == '__main__': | |
app.run(host='127.0.0.1', port=8001, debug=True) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment