Skip to content

Instantly share code, notes, and snippets.

@pgjones
Last active December 22, 2020 15:09
Show Gist options
  • Save pgjones/b6c062a0d04e2d45774f3a72625da071 to your computer and use it in GitHub Desktop.
Save pgjones/b6c062a0d04e2d45774f3a72625da071 to your computer and use it in GitHub Desktop.
DROP TABLE IF EXISTS todos;
CREATE TABLE todos (
id INTEGER PRIMARY KEY AUTOINCREMENT,
complete BOOLEAN NOT NULL DEFAULT FALSE,
due TIMESTAMPTZ,
task TEXT NOT NULL
);
import os
from sqlite3 import dbapi2 as sqlite3
from flask import Flask, request
app = Flask(__name__)
@app.route("/")
def index() -> str:
db = _create_db_connection()
result = db.execute("SELECT COUNT(*) as count FROM todos").fetchone()
return f"There are {result['count']} todos"
@app.route("/todos/", methods=["POST"])
def create_todo() -> tuple[dict, int]:
"""Create a new Todo.
This allows todos to be created and stored.
"""
data = request.get_json()
db = _create_db_connection()
cursor = db.execute(
"""INSERT INTO todos (complete, due, task)
VALUES (?, ?, ?)""",
[data["complete"], data["due"], data["task"]],
)
db.commit()
todo = data
todo["id"] = cursor.lastrowid
return todo, 201
@app.route("/todos/", methods=["GET"])
def get_todos() -> dict:
"""Get the todos.
Fetch all the Todos from the database.
"""
db = _create_db_connection()
if "complete" not in request.args:
result = db.execute(
"""SELECT id, complete, due, task
FROM todos""",
).fetchall()
else:
result = db.execute(
"""SELECT id, complete, due, task
FROM todos
WHERE complete = ?""",
[request.args["complete"]],
)
todos = [{**row} for row in result]
return {"todos": todos}
@app.route("/todos/<int:id>/", methods=["PUT"])
def update_todo(id: int) -> dict:
"""Update the identified todo
This allows the todo to be replace with the request data.
"""
data = request.get_json()
db = _create_db_connection()
db.execute(
"""UPDATE todos
SET complete = ?, due = ?, task = ?
WHERE id = ?""",
[data["complete"], data["due"], data["task"], id],
)
db.commit()
todo = data
todo["id"] = id
return todo
@app.route("/todos/<int:id>/", methods=["DELETE"])
def delete_todo(id: int) -> tuple[str, int]:
"""Delete the identified todo
This will delete the todo from the database.
"""
db = _create_db_connection()
db.execute(
"DELETE FROM todos WHERE id = ?",
[id],
)
db.commit()
return "", 202
def _create_db_connection():
engine = sqlite3.connect(os.path.join(app.root_path, "todos.db"))
engine.row_factory = sqlite3.Row
return engine
@app.cli.command("init_db")
def init_db() -> None:
db = _create_db_connection()
with app.open_resource("schema.sql", "r") as file_:
db.cursor().executescript(file_.read())
db.commit()
@pgjones
Copy link
Author

pgjones commented Dec 22, 2020

An example insert,

curl -vX POST -H "Content-Type: application/json" -d '{"complete": false, "due": "2021-01-01T00:00:00Z", "task": "tests"}' http://localhost:5000/todos/

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment