Skip to content

Instantly share code, notes, and snippets.

@arundhaj
Created January 23, 2022 12:08
Show Gist options
  • Save arundhaj/51bcb0df82b5f27f25b54bfe4afb7651 to your computer and use it in GitHub Desktop.
Save arundhaj/51bcb0df82b5f27f25b54bfe4afb7651 to your computer and use it in GitHub Desktop.
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import select, insert, update, delete
from sqlalchemy.exc import SQLAlchemyError
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///todo.sqlite3'
db = SQLAlchemy(app)
class Todo(db.Model):
TodoId = db.Column(db.Integer, primary_key = True)
Description = db.Column(db.String(255))
Status = db.Column(db.Boolean)
@app.route("/")
def hello_world():
return "Hello World!!!"
@app.route("/todo", methods=['GET', 'POST'])
def todos():
if request.method == 'GET':
select_todo_list = [Todo.TodoId, Todo.Description, Todo.Status]
select_todo_stmt = select(select_todo_list).select_from(Todo)
select_todo_result = db.session.execute(select_todo_stmt).fetchall()
todo_result_dict_list = [dict(todo) for todo in select_todo_result]
return jsonify(todo_result_dict_list)
elif request.method == 'POST':
try:
insert_todo_dict = request.json
insert_todo_dict['Status'] = False
insert_todo_stmt = insert(Todo).values(insert_todo_dict)
db.session.execute(insert_todo_stmt)
db.session.commit()
except SQLAlchemyError as e:
db.session.rollback()
raise e
return {
'Status': 'Successfully added the task'
}
@app.route("/todo/<int:todo_id>", methods=['GET', 'PUT', 'DELETE'])
def todo(todo_id):
if request.method == 'GET':
select_todo_list = [Todo.TodoId, Todo.Description, Todo.Status]
select_todo_stmt = select(select_todo_list).select_from(Todo).where(Todo.TodoId == todo_id)
select_todo_result = db.session.execute(select_todo_stmt).fetchone()
if select_todo_result is None:
return {
'Status': f'Todo with Id: {todo_id} not found'
}
select_todo_dict = dict(select_todo_result)
return select_todo_dict
elif request.method == 'PUT':
try:
update_todo_dict = request.json
update_todo_stmt = update(Todo).values(update_todo_dict).where(Todo.TodoId == todo_id)
db.session.execute(update_todo_stmt)
db.session.commit()
except SQLAlchemyError as e:
db.session.rollback()
raise e
return {
'Status': f'Todo with Id: {todo_id} updated successfully'
}
elif request.method == 'DELETE':
try:
delete_todo_stmt = delete(Todo).where(Todo.TodoId == todo_id)
db.session.execute(delete_todo_stmt)
db.session.commit()
except SQLAlchemyError as e:
db.session.rollback()
raise e
return {
'Status': f'Todo with Id: {todo_id} deleted successfully'
}
if __name__ == "__main__":
db.create_all()
app.run(debug=True)
flask
SQLAlchemy
flask-sqlalchemy
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment