Snapshot working source code before adding database migrations

Author: acsany

build-a-rest-api-frontend/source_code_working/app.py

@@ -0,0 +1,17 @@
+from flask import render_template
+
+import config
+from models import Person
+
+app = config.connex_app
+app.add_api(config.basedir / "swagger.yml")
+
+
+@app.route("/")
+def home():
+    people = Person.query.all()
+    return render_template("home.html", people=people)
+
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=8000, debug=True)

build-a-rest-api-frontend/source_code_working/build_database.py

@@ -0,0 +1,61 @@
+from datetime import datetime
+
+from config import app, db
+from models import Note, Person
+
+PEOPLE_NOTES = [
+    {
+        "lname": "Fairy",
+        "fname": "Tooth",
+        "notes": [
+            ("I brush my teeth after each meal.", "2022-01-06 17:10:24"),
+            (
+                "The other day a friend said, I have big teeth.",
+                "2022-03-05 22:17:54",
+            ),
+            ("Do you pay per gram?", "2022-03-05 22:18:10"),
+        ],
+    },
+    {
+        "lname": "Ruprecht",
+        "fname": "Knecht",
+        "notes": [
+            (
+                "I swear, I'll do better this year.",
+                "2022-01-01 09:15:03",
+            ),
+            (
+                "Really! Only good deeds from now on!",
+                "2022-02-06 13:09:21",
+            ),
+        ],
+    },
+    {
+        "lname": "Bunny",
+        "fname": "Easter",
+        "notes": [
+            (
+                "Please keep the current inflation rate in mind!",
+                "2022-01-07 22:47:54",
+            ),
+            ("No need to hide the eggs this time.", "2022-04-06 13:03:17"),
+        ],
+    },
+]
+
+with app.app_context():
+    db.drop_all()
+    db.create_all()
+    for data in PEOPLE_NOTES:
+        new_person = Person(lname=data.get("lname"), fname=data.get("fname"))
+        for content, timestamp in data.get("notes", []):
+            new_person.notes.append(
+                Note(
+                    content=content,
+                    timestamp=datetime.strptime(
+                        timestamp, "%Y-%m-%d %H:%M:%S"
+                    ),
+                )
+            )
+        db.session.add(new_person)
+    db.session.commit()

build-a-rest-api-frontend/source_code_working/config.py

@@ -0,0 +1,15 @@
+import pathlib
+
+import connexion
+from flask_marshmallow import Marshmallow
+from flask_sqlalchemy import SQLAlchemy
+
+basedir = pathlib.Path(__file__).parent.resolve()
+connex_app = connexion.App(__name__, specification_dir=basedir)
+
+app = connex_app.app
+app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{basedir / 'people.db'}"
+app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
+
+db = SQLAlchemy(app)
+ma = Marshmallow(app)

build-a-rest-api-frontend/source_code_working/models.py

@@ -0,0 +1,56 @@
+from datetime import datetime
+
+from marshmallow_sqlalchemy import fields
+
+from config import db, ma
+
+
+class Note(db.Model):
+    __tablename__ = "note"
+    id = db.Column(db.Integer, primary_key=True)
+    person_id = db.Column(db.Integer, db.ForeignKey("person.id"))
+    content = db.Column(db.String, nullable=False)
+    timestamp = db.Column(
+        db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow
+    )
+
+
+class NoteSchema(ma.SQLAlchemyAutoSchema):
+    class Meta:
+        model = Note
+        load_instance = True
+        sqla_session = db.session
+        include_fk = True
+
+
+class Person(db.Model):
+    __tablename__ = "person"
+    id = db.Column(db.Integer, primary_key=True)
+    lname = db.Column(db.String(32))
+    fname = db.Column(db.String(32))
+    timestamp = db.Column(
+        db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow
+    )
+
+    notes = db.relationship(
+        Note,
+        backref="person",
+        cascade="all, delete, delete-orphan",
+        single_parent=True,
+        order_by="desc(Note.timestamp)",
+    )
+
+
+class PersonSchema(ma.SQLAlchemyAutoSchema):
+    class Meta:
+        model = Person
+        load_instance = True
+        sqla_session = db.session
+        include_relationships = True
+
+    notes = fields.Nested(NoteSchema, many=True)
+
+
+note_schema = NoteSchema()
+person_schema = PersonSchema()
+people_schema = PersonSchema(many=True)

build-a-rest-api-frontend/source_code_working/notes.py

@@ -0,0 +1,50 @@
+from flask import abort, make_response
+
+from config import db
+from models import Note, Person, note_schema
+
+
+def read_one(note_id):
+    note = Note.query.get(note_id)
+
+    if note is not None:
+        return note_schema.dump(note)
+    else:
+        abort(404, f"Note with ID {note_id} not found")
+
+
+def update(note_id, note):
+    existing_note = Note.query.get(note_id)
+
+    if existing_note:
+        update_note = note_schema.load(note, session=db.session)
+        existing_note.content = update_note.content
+        db.session.merge(existing_note)
+        db.session.commit()
+        return note_schema.dump(existing_note), 201
+    else:
+        abort(404, f"Note with ID {note_id} not found")
+
+
+def delete(note_id):
+    existing_note = Note.query.get(note_id)
+
+    if existing_note:
+        db.session.delete(existing_note)
+        db.session.commit()
+        return make_response(f"{note_id} successfully deleted", 204)
+    else:
+        abort(404, f"Note with ID {note_id} not found")
+
+
+def create(note):
+    person_id = note.get("person_id")
+    person = Person.query.get(person_id)
+
+    if person:
+        new_note = note_schema.load(note, session=db.session)
+        person.notes.append(new_note)
+        db.session.commit()
+        return note_schema.dump(new_note), 201
+    else:
+        abort(404, f"Person not found for ID: {person_id}")

build-a-rest-api-frontend/source_code_working/people.py

@@ -0,0 +1,55 @@
+from flask import abort, make_response
+
+from config import db
+from models import Person, people_schema, person_schema
+
+
+def read_all():
+    people = Person.query.all()
+    return people_schema.dump(people)
+
+
+def create(person):
+    lname = person.get("lname")
+    existing_person = Person.query.filter(Person.lname == lname).one_or_none()
+
+    if existing_person is None:
+        new_person = person_schema.load(person, session=db.session)
+        db.session.add(new_person)
+        db.session.commit()
+        return person_schema.dump(new_person), 201
+    else:
+        abort(406, f"Person with last name {lname} already exists")
+
+
+def read_one(lname):
+    person = Person.query.filter(Person.lname == lname).one_or_none()
+
+    if person is not None:
+        return person_schema.dump(person)
+    else:
+        abort(404, f"Person with last name {lname} not found")
+
+
+def update(lname, person):
+    existing_person = Person.query.filter(Person.lname == lname).one_or_none()
+
+    if existing_person:
+        update_person = person_schema.load(person, session=db.session)
+        existing_person.fname = update_person.fname
+        db.session.merge(existing_person)
+        db.session.commit()
+        return person_schema.dump(existing_person), 201
+    else:
+        abort(404, f"Person with last name {lname} not found")
+
+
+def delete(lname):
+    existing_person = Person.query.filter(Person.lname == lname).one_or_none()
+
+    if existing_person:
+        db.session.delete(existing_person)
+        db.session.commit()
+        return make_response(f"{lname} successfully deleted", 200)
+    else:
+        abort(404, f"Person with last name {lname} not found")