# Demo {% hint style="danger" %} This demo is not up to date with Fall 2020 as we won't be covering DAO. We recommend you watch the demo video instead. {% endhint %} This demo will introduce you to a new python package, **SQLAlchemy**, which will help us interact with the database via an object relational mapping. We will also demonstrate how to set up a `dao.py` file to interface between our application and database code. ## File Setup Write the following code at the top of our `db.py` file: {% code title="db.py" %} ```python from flask_sqlalchemy import SQLAlchemy db = SQLAlchemy() ``` {% endcode %} We first need to import our `flask_sqlalchemy` package and instantiate the SQLAlchemy class. As the name implies, db will represent our database object that we will use to read and write to our database. Moving over to `app.py`, let’s first import our necessary models from `db.py` and initialize our application. {% code title="app.py" %} ```python import json from flask import Flask, request import dao from db import db # define db filename db_filename = "todo.db" app = Flask(__name__) # setup config app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{db_filename}" app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False app.config["SQLALCHEMY_ECHO"] = True # initialize app db.init_app(app) with app.app_context(): db.create_all() # generalized response formats def success_response(data, code=200): return json.dumps({"success": True, "data": data}), code def failure_response(message, code=404): return json.dumps({"success": False, "error": message}), code ``` {% endcode %} Recall that our `db` variable is an instance of the SQLAlchemy class, our database engine. To initialize our application, we will use the `init_app()` method that takes in a Flask application as an argument. Upon creating the application we will run `db.create_all()` to create all of our tables as defined by our models. ## Tasks ### Model Definition We define tables in our database just like a typical python class: {% code title="db.py" %} ```python class Task(db.Model): __tablename__ = "task" id = db.Column(db.Integer, primary_key=True) description = db.Column(db.String, nullable=False) done = db.Column(db.Boolean, nullable=False) ``` {% endcode %} #### **Class definition** For our first class, we will define the structure for our tasks. We will want to subclass off of the SQLAlchemy’s Model class to inherit all of the class properties needed to interact with this table. We also want to define the name of the table SQLAlchemy will create for us with the `__tablename__` property (note the double underscores). SQLAlchemy will autogenerate a table name for us if this property is not defined, but explicitly defining the table name can be helpful later on if we have to debug raw SQL commands and need to be sure of what tables correspond with what models. #### **Column definition** We define columns in our table via creating a new field in the class with db.Column. The column generation method takes in a mandatory argument to indicate the type of the column (all types found in [documentation](https://docs.sqlalchemy.org/en/13/core/type_basics.html)) and a series of optional arguments to define more characteristics of the column. The first column we wish to define is our unique identifier, id. We will be using an integer like before and SQLAlchemy provides a handy primary\_key=True flag to enforce the conditions we wish to set for a primary key, namely uniqueness and non-nullable. \ We define description and done with a string and boolean type respectively and also enforce that these fields are not nullable.\ \ Thinking ahead to how we will want to utilize the ORM for handling requests within our routes, we will need to convert an instance of the Task class into basic types. As mentioned prior, we need to send a standardized data type across HTTP requests (our particular choice is JSON), so let us define a method that helps us convert an object into a python dictionary. ```python def serialize(self): return { "id": self.id, "description": self.description, "done": self.done, } ``` This process of converting custom object types into the basic types is known as **serialization**. Our serialize method simply constructs a python dictionary from our object’s field values. ### 1. Get all tasks Our first method is quite basic but introduces us to the first use of our ORM. {% code title="app.py" %} ```python @app.route("/tasks/") def get_tasks(): tasks = [t.serialize() for t in Task.query.all()] return success_response(tasks) ``` {% endcode %} **Querying the database** Our first change is how we query all of our tasks. Before we had a method that used SQLite3 to write direct SQL queries on our database (in this case `SELECT * FROM task`). Now the ORM provides us a structure to query without directly writing any SQL! The `Task` class imported at the top gives us direct access to the `task` table. SQLAlchemy's `.query` reference enables us to write queries on our table and is a field for every SQLAlchemy defined class we will use. The `.all()` method functions exactly like a `SELECT * FROM task` by imposing no condition on which tasks are returned. **Constructing a return** Now because our task query call returns us **Task objects**, we need to serialize these into python dictionaries so that they are convertible to a JSON format. For this method and all others defined in the DAO, we want to provide `app.py` data that is as ready to be sent back into a response as possible which is why we will serialize it here. {% hint style="info" %} This one-line style of looping through an existing list is known as [**list comprehension**](https://www.pythonforbeginners.com/basics/list-comprehensions-in-python) and is highly recommended for simple loops like this. {% endhint %} ### 2. Create a task ```python @app.route("/tasks/", methods=["POST"]) def create_task(): body = json.loads(request.data) new_task = Task( description=body.get('description'), done=body.get('done', False) ) db.session.add(new_task) db.session.commit() return success_response(new_task.serialize(), 201) ``` **Construct Task object** First, we want to construct a python object to represent our task. Just like before, we need to parse our incoming request's body for the `description` data. We use python's built-in `.get()` method to safely parse the `body` dictionary. If there does not exist a value with the provided key, the method with return `None` by default. We can indicate a default value of `False` for the `done` variable using a second argument to `body.get()`. Next we need to use the ORM to convert this object into a real entry in the database. **Saving to the database** Note that we now have a `db` object which isolates the interaction with our database. Our `db` object has a field called `session` to keep track of our current connection to the database (remember how with sqlite3 we used `self.conn`). Just like with our sqlite3 implementation where we had separate commands for executing the insertion of a new task and the actual saving, or "**committing**", of a task, the same applies with our ORM. By keeping the addition of new entries to the database and actual committing, we free ourselves to potentially add multiple entries to different tables before accepting the cost of a database save. Now we can simply return a serialized task back to the client with a `201` response code to indicate the successful creation of a new entry. ### 3. Retrieve a task ```python @app.route("/tasks//") def get_task(task_id): task = Task.query.filter_by(id=task_id).first() if task is None: return failure_response("Task not found!") return success_response(task.serialize()) ``` Just like before, we design this route to accept an id within the route path. Now we can make use of SQLAlchemy's `filter_by` method to search for a task with the `id` field matching our provided `task_id`. This query will return all tasks that abide this criteria, functionally returning a **list** of tasks. However, we only want a single task so we call `.first()` at the end of our query to return the first instance returned in this query. This query will return `None` if there does not exist any tasks belonging to `task_id`. We return a failure response if the task does not exist in our table and otherwise return a success response with a serialized task. ### 4. Update a task ```python @app.route("/tasks//", methods=["POST"]) def update_task(task_id): task = Task.query.filter_by(id=task_id).first() if task is None: return failure_response("Task not found!") body = json.loads(request.data) task.description = body.get('description', task.description) task.done = body.get('done', task.done) db.session.commit() return success_response(task.serialize()) ``` First, we want to query for our task just like before and return if there does not exist said task. After parsing the request's body, we can update our `task` object's fields like a normal python custom object with direct assignment to the fields `task.description` and `task.done`. We get a bit fancy with our use of `.get()` by placing the current `description` or `done` value as the default if no new value is provided in the request. Therefore, absent fields in the body will not change our database. Upon assigning new values to the task, we need to commit these changes and return our serialized object. ### 5. Delete a task ```python @app.route("/tasks//", methods=["DELETE"]) def delete_task(task_id): task = Task.query.filter_by(id=task_id).first() if task is None: return failure_response("Task not found!") db.session.delete(task) db.session.commit() return success_response(task.serialize()) ``` First, we want to query for our task just like before and return if there does not exist said task. We can use SQLAlchemy's provided `delete()` method to remove our task from the database. Upon committing our changes, we can return the original task to give back to the client. ## Subtasks (One-to-Many Relationship) ### Model Definition Consider that we want to add smaller “subtasks” within each task to break it down to more manageable components. We can begin defining a Subtask model with a One-to-Many relationship with Task as follows: ```python class Subtask(db.Model): __tablename__ = "subtask" id = db.Column(db.Integer, primary_key=True) description = db.Column(db.String, nullable=False) done = db.Column(db.Boolean, nullable=False) task_id = db.Column(db.Integer, db.ForeignKey("task.id"), nullable=False) ``` We follow our same guidelines of defining a `__tablename__` field and all of our columns. But to create the relationship between a Task and Subtask, we do something special. We define task\_id as a column in the table and give it the type of an integer, but we also add the argument `db.ForeignKey("task.id")`. This part indicates the declaration of a foreign key that references the id field in the task table. To generalize this structure, we define foreign keys with: `db.ForeignKey(".")`. **Serialization** Our next step is to define how we wish to serialize a Subtask and how this serialization will interact with our existing serialization for Task. We can first write a straightforward serialization of a Subtask: ```python def serialize(self): return { "id": self.id, "description": self.description, "done": self.done } ``` #### ### Update Task Model The second part to defining a one-to-many relationship involves adding some indicator code to the model on the other side of the relationship. Within our Task model we now write: ```python class Task(db.Model): __tablename__ = "task" id = db.Column(db.Integer, primary_key=True) description = db.Column(db.String, nullable=False) done = db.Column(db.Boolean, nullable=False) # defining the reverse side of the relationship subtasks = db.relationship("Subtask", cascade="delete") ``` We create a subtasks field that allows us to query from the reverse side of the relationship by defining a db.relationship type. The first argument defines the name of the model class we are referencing (notice how this is `Subtask` instead the `subtask` table name). Our second and optional argument, cascade='delete', defines a protocol to handle a deletion of a Task object from the database. Consider we remove a Task with many Subtasks, all of these subtasks now have no parent Task to reference. By default, their `task_id` field will be set to None, but in the context of our app, we only want Subtasks to exist as nested beneath a main Task. The cascade protocol says to the ORM to delete all related Subtasks upon deleting a Task to preserve this nature of the application. #### Update Task Serializer Now let's consider how we want to access these subtasks from the client’s perspective. We have two options: add a new route for the client to request for the subtasks of a particular task **or** find a way to return subtask information in the original requests that retrieve Tasks. We will go with the latter due to its simplicity for us and the client. Let us update our Task serializer to include a new field:
```python def serialize(self): return { "id": self.id, "description": self.description, "done": self.done, # serialize each subtask related to this task "subtasks": [s.serialize() for s in self.subtasks], } ``` We can use the list comprehension approach as before to serialize all of our subtasks. Our original list, self.subtasks, utilizes our defined relationship to actually make a query and gather our related subtasks. We can then serialize each of these objects in our Task serializer method. ### 6. Create a subtask {% code title="app.py" %} ```python @app.route("/tasks//subtasks/", methods=["POST"]) def create_subtask(task_id): task = Task.query.filter_by(id=task_id).first() if task is None: return failure_response("Task not found!") body = json.loads(request.data) new_subtask = Subtask( description=body.get('description'), done=body.get('done'), task_id=task_id ) db.session.add(new_subtask) db.session.commit() return success_response(new_subtask.serialize()) ``` {% endcode %} We will create a new subtask by instantiating the `Subtask` class, just like we did for regular tasks. To complete the addition, we add the newly created subtask to our database, commit the changes and finally return the subtask serialized. Upon updating our models' structure, we can implement a new route to add subtasks to a parent task: Route Definition: When creating new subtasks, we want to provide clients with a means to indicate which task to attach it with. Of the many ways for us to do this, we will simply extend our traditional `task/task_id/` route structure by adding a `subtask/` to the route path. Because we are expecting data to come in our request, we want to accept the `POST` request method. Subtask Addition: First, we need to retrieve the task we intend to add this new subtask under and return a failed response if we cannot find it. Then, we can move forward with parsing our request's body, calling our DAO, and returning a success response. ## Categories (Many-to-Many Relationship) ### Model Definition We will demonstrate a many-to-many relationship with a categories feature. Any given task can belong to many different category groups and a category can contain many tasks associated with it. We can define a basic implementation as follows: {% code title="db.py" %} ```python class Category(db.Model): __tablename__ = "category" id = db.Column(db.Integer, primary_key=True) description = db.Column(db.String, nullable=False) color = db.Column(db.String, nullable=False) ``` {% endcode %} **Serialization** Our next step is to define the serialization of a `Category` with an analogous implementation to our previous models: ```python def serialize(self): return { "id": self.id, "description": self.description, "color": self.color } ``` ### Create Association Table As of right now, our `Category` model is not related to anything. Recall from our Relational Databases lecture that we implement a many-to-many relationship via a "join" or "association" table to connect the foreign key of one table to the foreign key another table. At the top of our `db.py` file, we can create a new variable to construct a table using `db.Table`: ```python association_table = db.Table("association", db.Model.metadata, db.Column("task_id", db.Integer, db.ForeignKey("task.id")), db.Column("category_id", db.Integer, db.ForeignKey("category.id")) ) ``` The first argument provided is the table name, which we will call `"association"`. Next we need to provide the metadata of the `Model` class. All following arguments will be the columns of our table. Our first column we will store the ids of tasks, `db.ForeignKey` type, and our second column will store the ids of categories, also `db.ForeignKey` type. ### Add Relationship Fields Now to implement this relationship, we connect our `Category` model to our association table with a `db.relationship` field: ```python class Category(db.Model): __tablename__ = "category" id = db.Column(db.Integer, primary_key=True) description = db.Column(db.String, nullable=False) color = db.Column(db.String, nullable=False) # define many-to-many relationship by connecting to association table tasks = db.relationship("Task", secondary=association_table, back_populates='categories') ``` The first argument indicates the **class name** of the model we are relating too, not the table name which is why `"Task"` is capitalized. Our second argument, `secondary`, indicates the join table to use. Finally, `back_populates` indicates to SQLAlchemy the field name on the reverse side of the relationship that will be referencing this relationship. We can see this in our addition to the `Task` model: ```python class Task(db.Model): __tablename__ = "task" id = db.Column(db.Integer, primary_key=True) description = db.Column(db.String, nullable=False) done = db.Column(db.Boolean, nullable=False) # defining the reverse side of the many-to-one relationship subtasks = db.relationship("Subtask", cascade="delete") # defining the reverse side of the many-to-many relationship, connecting to the same join table categories = db.relationship("Category", secondary=association_table, back_populates="tasks") ``` We define the new field `categories` to match the input of `back_populates` from the relationship definition in the `Category` model. We also set the `back_populates` on this side to the field name we previously defined: `"tasks"`. Our first argument, the class name, will refer to the `Category` model, and we will refer to the same `association_table`. **It is crucial that the `back_populates` tags match exactly with the field names across the two models.** #### Update Task Serializer Just like with subtasks, we want to serialize categories within a task. We can do this analogously with another new line in our `Task` serializer: ```python def serialize(self): return { "id": self.id, "description": self.description, "done": self.done, # serialize each subtask related to this task "subtasks": [s.serialize() for s in self.subtasks], # serialize each category related to this task "categories": [c.serialize() for c in self.categories] } ``` ### 7. Assign category to task ```python def assign_category(task_id, description, color): task = get_task_by_id(task_id) if task is None: return None category = _get_or_create_category(description, color) task.categories.append(category) db.session.commit() return task.serialize() ``` First, we need to retrieve the task we intend to link to our category and return `None` if we cannot find it so our route knows to return a failure response. We then can call our custom method to get or create a category from the provided description. To relate this category to our task, we can append it to the `task.categories` field. Relationship fields in SQLAlchemy can be manipulated as lists which is why we use the `.append()` method to add to our relationship. Doing so will automatically update a new entry in the previously defined association table between the two models. Now we can commit changes and return the serialized task. The request body is expected to contain a `description` field and we will try to find any categories that match this description (we will not be allowing categories of the same description to exist). If there does not exist a category with said description, we will create a new `Category` object with the description and color from the request's body. This method is **private** and only intended to be used within other DAO methods (for us, our next method to assign categories). Private methods in python are indicated with a prepended underscore `_`. Just like with subtasks, we want to add a new route for assigning a category to a task: ```python @app.route("/tasks//category/", methods=["POST"]) def assign_category(task_id): body = json.loads(request.data) task = dao.assign_category(task_id, body.get("description"), body.get("color"),) if task is None: return failure_response("Task not found!") return success_response(task) ``` Following previous design patterns, we will extend our usual path for a specific task with the `category/` string and make this route accept the `POST` request method. Because we off-loaded the important logic to our DAO, our route method remains relatively simple and clean to read.