Last active
December 22, 2020 15:09
-
-
Save pgjones/b6c062a0d04e2d45774f3a72625da071 to your computer and use it in GitHub Desktop.
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
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 | |
); |
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
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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
An example insert,