app.py

import random
from sqlite3 import Connection as SQLite3Connection
from datetime import datetime
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import event
from sqlalchemy.engine import Engine
from flask import Flask, request, jsonify

from DataStructures.BinarySearchTree import BinarySearchTree
from DataStructures.LinkedList import LinkedList
from DataStructures.HashTable import HashTable
from DataStructures.Queue import Queue
from DataStructures.Stack import Stack

app = Flask(__name__)

app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///sqlitedb.file"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = 0


# configure sqlite3 to enfore foreign key constraints
# without this config, sqlite would allow blog posts without valid users
@event.listens_for(Engine, "connect")
def _set_sqlite_pragma(dbapi_connection, connection_record):
    if isinstance(dbapi_connection, SQLite3Connection):
        cursor = dbapi_connection.cursor()
        cursor.execute("PRAGMA foreign_keys=ON;")
        cursor.close()


# get an instance of the SQLAlchemy class for app
db = SQLAlchemy(app)
now = datetime.now()


# models
class User(db.Model):
    __tablename__ = "user"
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
    email = db.Column(db.String(50))
    address = db.Column(db.String(200))
    phone = db.Column(db.String(50))
    posts = db.relationship("BlogPost", cascade="all, delete")


class BlogPost(db.Model):
    __tablename__ = "blog_post"
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(50))
    body = db.Column(db.String(200))
    date = db.Column(db.Date)
    user_id = db.Column(db.Integer, db.ForeignKey("user.id"), nullable=False)


@app.route("/user", methods=["POST"])
def create_user():
    data = request.get_json()
    new_user = User(
        name=data["name"],
        email=data["email"],
        address=data["address"],
        phone=data["phone"]
    )
    db.session.add(new_user)
    db.session.commit()
    return jsonify({"message": "User created"}), 200


@app.route("/users/descending", methods=["GET"])
def get_all_users_descending():
    users = User.query.all()
    all_users = LinkedList.LinkedList()

    for user in users:
        all_users.insert_head(
            {
                "id": user.id,
                "name": user.name,
                "email": user.email,
                "address": user.address,
                "phone": user.phone
            }
        )

    return jsonify(all_users.to_list()), 200


@app.route("/users/ascending", methods=["GET"])
def get_all_users_ascending():
    users = User.query.all()
    all_users = LinkedList.LinkedList()

    for user in users:
        all_users.insert_tail(
            {
                "id": user.id,
                "name": user.name,
                "email": user.email,
                "address": user.address,
                "phone": user.phone
            }
        )

    return jsonify(all_users.to_list()), 200


@app.route("/user/<user_id>", methods=["GET"])
def get_one_user(user_id):
    users = User.query.all()
    all_users = LinkedList.LinkedList()

    for user in users:
        all_users.insert_head(
            {
                "id": user.id,
                "name": user.name,
                "email": user.email,
                "address": user.address,
                "phone": user.phone
            }
        )

    user = all_users.get_user_by_id(user_id)

    return jsonify(user), 200


@app.route("/user/<user_id>", methods=["DELETE"])
def delete_user(user_id):
    user = User.query.filter_by(id=user_id).first()
    db.session.delete(user)
    db.session.commit()
    return jsonify({}), 200


@app.route("/blog_post/<user_id>", methods=["POST"])
def create_blog_post(user_id):
    data = request.get_json()

    user = User.query.filter_by(id=user_id).first()
    if not user:
        return jsonify({"message": "user does not exist"}), 400

    ht = HashTable.HashTable(10)
    ht.add_key_value("title", data["title"])
    ht.add_key_value("body", data["body"])
    ht.add_key_value("date", now)
    ht.add_key_value("user_id", user_id)

    new_blog_post = BlogPost(
        title=ht.get_value("title"),
        body=ht.get_value("body"),
        date=ht.get_value("date"),
        user_id=ht.get_value("user_id")
    )

    db.session.add(new_blog_post)
    db.session.commit()
    return jsonify({"message": "Blog Post created"}), 200


@app.route("/blog/<blog_post_id>", methods=["GET"])
def get_all_blog_posts(blog_post_id):
    posts = BlogPost.query.all()
    random.shuffle(posts)

    bst = BinarySearchTree.BinarySearchTree()

    for post in posts:
        bst.insert({
            "id": post.id,
            "title": post.title,
            "body": post.body,
            "user_id": post.user_id
        })

    post = bst.search(blog_post_id)

    if not post:
        return jsonify({"message": "Post not found"})

    return jsonify(post), 200


@app.route("/blog_post/numeric_body", methods=["GET"])
def get_numeric_post_bodies():
    posts = BlogPost.query.all()

    q = Queue.Queue()

    for post in posts:
        q.enqueue(post)

    return_list = []

    for _ in range(len(posts)):
        post = q.dequeue()
        numeric_body = 0
        for char in post.data.body:
            numeric_body += ord(char)

        post.data.body = numeric_body
        return_list.append(
            {
                "id": post.data.id,
                "title": post.data.title,
                "body": post.data.body,
                "user_id": post.data.user_id
            }
        )

    return jsonify(return_list), 200


@app.route("/blog_post/delete_last_10", methods=["DELETE"])
def delete_blog_post():
    posts = BlogPost.query.all()

    s = Stack.Stack()

    for post in posts:
        s.push(post)

    for _ in range(10):
        post_delete = s.pop()
        db.session.delete(post_delete.data)
        db.session.commit()

    return jsonify({"message": "success"}), 200


if __name__ == '__main__':
    app.run(debug=True)