Everything You Need to Know to Map SQL Tables to REST Endpoints in Python
The backbone of back-end programming is the act of sending, receiving, and manipulating data. The two most common ways of interacting with backend data are SQL (“Structured Query Language”) and REST (“REpresentational State Transfer”) APIs. SQL is the method that many servers use to store their data, while REST is a method of transferring that data to and from the client. Of course, there are many ways to store and send data, but these are the most common and are widely supported.
To start working with SQL and REST, the first thing you need to understand is how they actually integrate together.
To illustrate this, let’s imagine you’re an event business; there’s lots of data that you need to store and access. For example, you have events, bills for those events, and people attending those events and paying those bills. This blog explains the different kinds of relationships between pieces of data in SQL and how to use the four REST verbs to create a REST API to give users the ability to interact with the data stored in SQL.
- Introduction to SQL relationships
- Introduction to REST API design
- How to represent relationships in a REST API
- All API Endpoints
Introduction to SQL relationships
SQL is just a fancy way of saying it’s a database that stores data in tables (rows and columns, not things to eat off of!).
One of the many common themes in programming that SQL emphasizes is avoiding duplication of data. After all, who wants to have to change a field in multiple places when only one piece of data changed? Let’s min/max the effort spent.
In order to accomplish this, SQL needs a way to say that certain pieces of data are ‘related’ to other pieces of data. Putting all the data in one table would be rather unwieldy! However, there still needs to be a way to associate different pieces of data – like how you need to know the syntax for a programming language, but it’s useless if you don’t know how to compile or interpret the language.
Hence, there are three types of “relationships” in SQL: one to one, one to many, and many to many.
One to one
As the name implies, one to one means that there is only one of a certain thing related to another thing. For example, one event has only one bill (in a perfect world), and a bill is only for one event.
One to many
Here, one piece of data is related to multiple pieces of data of the same type. For example, one person can have many bills, but those bills are only meant to be paid by one person. Or, one cat may have nine lives, but each of those lives belongs to only one cat.
Many to many
Here, events can have multiple people attend them, and people can attend multiple events. For another example, a code repository can have multiple contributors, and each contributor can have multiple code repositories. Imagine if you could only have one!
The next question, though, is how do we store a many to many relationship without repeating data? The answer is that you can’t completely. But the next best thing is only having minimal data duplication. At first, you may think to make a list of IDs of related data, but lists are difficult to edit in SQL. So, the idiomatic solution for SQL is ‘junction’ tables!
A junction table typically has a name that is the names of two tables that have a many-to-many relationship squished together, and consists of foreign keys (meaning they reference the primary keys in other tables). So, for people and events the junction table would be people_events.
All of this means that every row in the table represents a person id and an event id, and multiple people can be signed up for one event and multiple events can be attended by one person. This leaves some duplication of data, but it is minimal. The table would look like:
-- create tables (people, events, bills, people_events)
create table people
(
id SERIAL,
name varchar(255),
primary key(id)
);
create table events
(
id SERIAL,
date DATE NOT NULL DEFAULT CURRENT_DATE,
type varchar(255),
primary key(id)
);
create table bills
(
id SERIAL,
cost int,
paid varchar(255) default 'not paid',
due DATE NOT NULL DEFAULT CURRENT_DATE,
payer_id int,
event_id int,
primary key(id),
foreign key (payer_id) references people(id),
foreign key (event_id) references events(id)
);
create table people_events
(
people_id int,
event_id int,
foreign key (people_id) references people(id),
foreign key (event_id) references events(id)
);
-- populate with sample data
insert into people (name) values ('PersonA'), ('PersonB'), ('PersonC');
insert into events (date, type)
values ('2023-02-20', 'concert'), ('2022-06-30', 'conference'), ('2025-12-20', 'holiday party');
insert into bills (cost, paid, due, payer_id, event_id) values (2000, 'paid', '2022-02-20', 1, 1), (10000, 'not paid', '2021-06-01', 1, 2), (5000, 'paid', '2025-01-20', 2, 3);
insert into people_events (people_id, event_id) values (1, 1), (1, 2), (2, 3), (2, 1);Code language: SQL (Structured Query Language) (sql)Now that we have our tables (people, bills, events, and people_events) the next step is making them usable without having to actually connect to a database. After all, we want to control what someone can change in the database, give ourselves a space to write business logic, and give users an easier interface to use our database with. The ideal way to do this is to map our database’s functionality to API (Application Programming Interface) endpoints.
Introduction to REST API design
One type of API is REST, which stands for Representational State Transfer. REST is a standard for how to change the state of data. REST defines an ‘endpoint’ as an url (just like the one you navigated to in order to read this blog) and a ‘verb’, which is what changes the data is experiencing.
There are four main REST verbs:
- GET: Literally means we ‘get’ the requested resource (data). You performed a “GET” just to get here.
- POST: This is how new data is created. We ‘post’ data to the API (like how we ‘post’ mail or comments).
- PATCH: Also sometimes called “PUT”, this is how data that exists is edited. It’d be a pain if we had to delete data and re-’post’ the data every time something changes, so we use ‘patch’ instead. To understand how annoying it could be without ‘patch’, imagine if the day of the week was a value in a table, and ergo you had to DELETE and POST it every day.
- DELETE: This is how existing data is removed.
With these four verbs we can provide our users the full CRUD (create, read, update, delete) functionality via the API to interact with our database tables.
But wait. If we give an API the URL /people and the verb POST (or PATCH)… how does the API know what data about the person to put in the database? Does it randomly generate a row in the table? That wouldn’t make much sense. Instead, the information about the person or other resource to create is given in the ‘body’ of the request. Typically this data is given in the form of JSON, and looks a bit like this:
{
"name":"New Person"
}Code language: JSON / JSON with Comments (json)POST and PATCH are the two verbs that expect a ‘body’ in their request. Because of this you should also pass in a header (“Content-type”) telling the API what type of data you’re passing in (in this case, since we’re returning JSON, it would be “Content-type: application/json”). The content type header is optional, and many APIs do not require it, but it is good practice.
An example of some REST headers would be:
HTTP/1.1 200 OK
Date: Fri, 08 Jul 2022 22:54:27 GMT
Content-type: application/json
Connection: keep-alive
Vary: Accept-Encoding, Origin
Content-Encoding: gzipCode language: HTTP (http)Grab and create data using GET and POST
When we’re thinking about how to design the API it should be noted that APIs are intended to be how programs talk to each other (meaning they’re intended for code such as UIs to call them, not for people to call them manually). This means that the priority when designing an API is not being easy for humans to manually use, but, instead, being easy for programs to interact with. Of course, when both can be achieved that is preferable.
The most straightforward way to start designing an API is to allow endpoints for each REST verb for each table (so the user has full CRUD access to each table).
In our case that means that for the table ‘people’ we start by creating the endpoint /people with the verb GET (so the user can “GET” all the people). This is called a ‘collection’ endpoint because it returns one or more resources (in this case, one resource is one person).
And in order to create new people (in the database, not real life!), the /people URL with the verb POST endpoint comes next.
Please note that, for this blog’s purposes, we’ll be using a Python based tool called Flask to create our API and a Python package called SQLAlchemy to access our database.
import sqlalchemy
from sqlalchemy.orm import sessionmaker
# from flask import Flask, Response, request
from typing import Dict
# app = Flask(__name__)
# Connect to the DB and reflect metadata.
engine = sqlalchemy.create_engine("postgresql://coderpad:@/coderpad?host=/tmp/postgresql/socket")
connection = engine.connect()
Session = sessionmaker(bind=engine)
session = Session()
metadata = sqlalchemy.MetaData()
metadata.reflect(bind=engine)
# @app.route("/people", methods=["GET"])
def get_all() -> Dict:
people_table = metadata.tables["people"]
result = connection.execute(people_table.select()).fetchall()
return {
"result": [
dict(person) for person in result
]
}
# @app.route("/people", methods=["POST"])
def post() -> Response:
content_type = request.headers.get("Content-Type")
if (content_type != "application/json"):
return {
"result": "Failure",
"reason": "Content-Type not supported!",
}
people_table = metadata.tables["people"]
# There should be validation that the POST body is valid here!
connection.execute(
sqlalchemy.insert(people_table).values(
**request.json,
)
)
return Response(status=201) # Created
# if __name__ == "__main__":
# app.run()Code language: Python (python)PATCH your data to update values
PATCH is the next REST verb we’ll touch on. PATCH gives users the ability to change data about resources. While we coulduse PATCH for the ‘/people’ endpoint – which would enable users to make “bulk” changes – let’s focus on the use-case of modifyingone“person” at a time.
When considering if you want bulk endpoints, the question is “does the user want to be able to change multiple rows in your table at a time?”. If yes, then you want bulk endpoints; if not, then this endpoint is extraneous.
To change one resource, then, we need a way to tell the API what row in the table to change. There are multiple ways to do that (such as query parameters), but the standard way of doing this is by putting the ID (or other unique identifier) of the resource in the url.
To update the first person created in the table, the endpoint would be /people/1 with the verb PATCH. This endpoint would update the person with the ID 1 using whatever information is passed in the ‘body’ of the request.
When documenting APIs this is usually shown by specifying the unique identifier using :attributeName or, in the case of Flask <datatype: attributeName>. In the Flask way of doing things, our PATCH endpoint would be /people/<int: id> with the verb PATCH.
import sqlalchemy
from sqlalchemy.orm import sessionmaker
from flask import Flask, Response, request
app = Flask(__name__)
# Connect to the DB and reflect metadata.
engine = sqlalchemy.create_engine("postgresql://coderpad:@/coderpad?host=/tmp/postgresql/socket")
connection = engine.connect()
Session = sessionmaker(bind=engine)
session = Session()
metadata = sqlalchemy.MetaData()
metadata.reflect(bind=engine)
@app.route("/people/<int:id>", methods=["PATCH"])
def patch(id: int) -> Response:
content_type = request.headers.get("Content-Type")
if (content_type != "application/json"):
return {
"result": "Failure",
"reason": "Content-Type not supported!",
}
people_table = metadata.tables["people"]
# There should be validation that the PATCH body is valid here!
connection.execute(
sqlalchemy.update(people_table).where(
people_table.c.id == id
).values(
**request.json,
)
)
return Response(status=204) # No Content
if __name__ == "__main__":
app.run()Code language: Python (python)How to DELETE your data
And when we get to the DELETE verb (to add the ability to delete people) we have the option to make a bulk endpoint. But, for brevity’s sake, we’ll focus on deleting one person at a time, using the endpoint /people/<int: id> with the verb DELETE.
import sqlalchemy
from sqlalchemy.orm import sessionmaker
from flask import Flask, Response
app = Flask(__name__)
# Connect to the DB and reflect metadata.
engine = sqlalchemy.create_engine("postgresql://coderpad:@/coderpad?host=/tmp/postgresql/socket")
connection = engine.connect()
Session = sessionmaker(bind=engine)
session = Session()
metadata = sqlalchemy.MetaData()
metadata.reflect(bind=engine)
@app.route("/people/<int:id>", methods=["DELETE"])
def delete(id: int) -> Response:
people_table = metadata.tables["people"]
connection.execute(
sqlalchemy.delete(people_table).where(
people_table.c.id == id
)
)
return Response(status=204) # No Content
if __name__ == "__main__":
app.run()Code language: Python (python)To summarize, the four endpoints you’re most likely to have in an API per (non-junction) table are:
`/people` with GET,
`/people` with POST,
`/people/<int: id>` with PATCH, and
`/people/<int: id>` with DELETECode language: JavaScript (javascript)And the same four endpoints (with a different url based on the table name) for each of our other (non-junction) tables: bills and events.
Notice anything strange? They all are /people! Why? URLs are broken down into several parts – the IP (internet protocol) which is, in a browser, going to be http or https, the domain (which is where the API is hosted – like how www.google.com is the domain for Google’s main page), then the API endpoint (after a /). If you turn your eyes to the URL in your browser right now you should see at least https://coderpad.io/blog/, meaning the endpoint is /blog/ and a GET verb.
Because of this, the way endpoints are documented uses just the URL (sometimes called a path or route) and the verb because they are unique to our API, whereas we could have our API hosted in multiple places on different protocols.
These four endpoints are a great start to understanding REST APIs, but from here you may start to think of other things you want the API to do. For example, what if you only want information on one person? To GET only one person (one resource) we employ the same tactic of putting the ID in the URL, and we end up with /people/<int: id> with the GET verb.
Here you could change the response to not be a list since you’re always only returning one resource. However, it can be nice for it to remain a list because then the ‘instance’ (returning only one resource) endpoint returns in the same format as the ‘collection’ (returning one or more resources) endpoint, and because of this consistency, the client’s code can reuse the same parsing functions.
import sqlalchemy
from sqlalchemy.orm import sessionmaker
from flask import Flask
from typing import Dict
app = Flask(__name__)
# Connect to the DB and reflect metadata.
engine = sqlalchemy.create_engine("postgresql://coderpad:@/coderpad?host=/tmp/postgresql/socket")
connection = engine.connect()
Session = sessionmaker(bind=engine)
session = Session()
metadata = sqlalchemy.MetaData()
metadata.reflect(bind=engine)
@app.route("/people/<int:id>", methods=["GET"])
def get_one(id: int) -> Dict:
people_table = metadata.tables["people"]
result = connection.execute(
people_table.select().where(people_table.c.id == id)
).fetchall()
return {
"result": [
dict(person) for person in result
]
}
if __name__ == "__main__":
app.run()Code language: Python (python)How to represent relationships in a REST API
So far we’ve covered how to give users create, read, update, and delete access to data in all our non-junction tables, but we haven’t covered how to change the relationships between the data. Here’s where APIs get a little more complex, because there is no longer a one table to four (one per verb) endpoints mapping.
Many-to-many relationships require dynamic routes
What if you want to know who has signed up for what event? This is a many-to-many relationship (as many people can be signed up for one event, and one person can be signed up for many events) in the junction table, people_events, so, to view or change data in many-to-many relationships we need more endpoints.
However, an API endpoint that is people_events (with any verb) seems a bit odd. Even if the ‘user’ is the engineer writing the code to call the API rather than a customer, the user probably doesn’t want to know there’s a table called people_events! And the machine definitely doesn’t either. Junction table names are magic that should stay behind the scenes. We want to make the API as intuitive as possible; even if it is intended for machines.
To have these intuitive endpoints, the standard is to add to the existing paths (the URL portion of our endpoints), giving people a ‘path’ to the related resources. So, for seeing which events someone is signed up for we recycle the path to one person /people/<int: id> and add /events and /events/<int: id> onto the end. Then we have both a collection and an instance endpoint.
/people/<int: people_id>/events with GET
/people/<int: people_id>/events/<int: event_id> with GETCode language: HTML, XML (xml)from typing import Dict
import sqlalchemy
from sqlalchemy.orm import sessionmaker
from flask import Flask
app = Flask(__name__)
# Connect to the DB and reflect metadata.
engine = sqlalchemy.create_engine("postgresql://coderpad:@/coderpad?host=/tmp/postgresql/socket")
connection = engine.connect()
Session = sessionmaker(bind=engine)
session = Session()
metadata = sqlalchemy.MetaData()
metadata.reflect(bind=engine)
@app.route("/people/<int:people_id>/events", methods=["GET"])
def get_people_events_all(people_id: int) -> Dict:
people_events_table = metadata.tables["people_events"]
events_table = metadata.tables["events"]
result = connection.execute(
events_table.select().join(
people_events_table,
events_table.c.id == people_events_table.c.event_id
).where(
people_events_table.c.people_id == people_id
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
@app.route("/people/<int:people_id>/events/<int:event_id>", methods=["GET"])
def get_people_events_one(people_id: int, event_id: int) -> Dict:
people_events_table = metadata.tables["people_events"]
events_table = metadata.tables["events"]
result = connection.execute(
events_table.select().join(
people_events_table,
events_table.c.id == people_events_table.c.event_id
).where(
(people_events_table.c.people_id == people_id)
& (people_events_table.c.event_id == event_id)
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
if __name__ == "__main__":
app.run()Code language: Python (python)Now the user can see many to many relationships, so the next step is enabling them to edit these relationships. For this type of endpoint, PATCH won’t accept a request ‘body’ and will instead expect /people/<int: people_id>/events/<int: event_id> where the event_id is the ID of an existing event. Then the API will create a relationship where the given person is signed up for the given event. Similarly, DELETE expects /people/<int: people_id>/events/<int: event_id> and only deletes the relationship between the person and the event.
Sometimes you see a POST verb which both creates the resource and the relationship, but it isn’t needed as the same functionality can be achieved by PATCH and POSTing to the /events URL, so for this blog we’ll talk about just PATCH and DELETE.
So all of our sub-endpoints for people would look like:
/people/<int: id>/events with GET
/people/<int: id>/events/1 with GET, PATCH, and DELETECode language: HTML, XML (xml)while the other side of the relationship, events would have the endpoints:
/events/<int: event_id>/people/ with GET
/events/<int: event_id>/people/<int: people_id> with GET, PATCH, and DELETECode language: HTML, XML (xml)In reality, the instance endpoints provide some redundant functionality – because relationships go both ways, one side of the relationship’s URLs could be removed and it would be fine. Meaning that
/people/<int: people_id>/events/<int: event_id> with GET, PATCH, and DELETECode language: HTML, XML (xml)provides the functionality that can be acquired from
/events/<int: event_id>/people/<int: people_id> with GET, PATCH, and DELETECode language: HTML, XML (xml)so only one of the two is needed, but all are included here for completeness’s sake.
from typing import Dict
import sqlalchemy
from sqlalchemy.orm import sessionmaker
from flask import Flask, Response
app = Flask(__name__)
# Connect to the DB and reflect metadata.
engine = sqlalchemy.create_engine("postgresql://coderpad:@/coderpad?host=/tmp/postgresql/socket")
connection = engine.connect()
Session = sessionmaker(bind=engine)
session = Session()
metadata = sqlalchemy.MetaData()
metadata.reflect(bind=engine)
def signup_user(people_id: int, event_id: int) -> None:
people_events_table = metadata.tables["people_events"]
connection.execute(
sqlalchemy.insert(people_events_table).values(
people_id=people_id,
event_id=event_id,
)
)
def delete_signup(people_id: int, event_id: int) -> None:
people_events_table = metadata.tables["people_events"]
connection.execute(
sqlalchemy.delete(people_events_table).where(
(people_events_table.c.people_id == people_id)
& (people_events_table.c.event_id == event_id)
)
)
@app.route("/people/<int:people_id>/events", methods=["GET"])
def get_people_events_all(people_id: int) -> Dict:
people_events_table = metadata.tables["people_events"]
events_table = metadata.tables["events"]
result = connection.execute(
events_table.select().join(
people_events_table,
events_table.c.id == people_events_table.c.event_id
).where(
people_events_table.c.people_id == people_id
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
@app.route("/people/<int:people_id>/events/<int:event_id>", methods=["GET"])
def get_people_events_one(people_id: int, event_id: int) -> Dict:
people_events_table = metadata.tables["people_events"]
events_table = metadata.tables["events"]
result = connection.execute(
events_table.select().join(
people_events_table,
events_table.c.id == people_events_table.c.event_id
).where(
(people_events_table.c.people_id == people_id)
& (people_events_table.c.event_id == event_id)
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
@app.route("/events/<int:event_id>/people", methods=["GET"])
def get_events_people_all(event_id: int) -> Dict:
people_events_table = metadata.tables["people_events"]
people_table = metadata.tables["people"]
result = connection.execute(
people_table.select().join(
people_events_table,
people_table.c.id == people_events_table.c.event_id
).where(
people_events_table.c.event_id == event_id
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
@app.route("/events/<int:event_id>/people/<int:people_id>", methods=["GET"])
def get_events_people_one(event_id: int, people_id: int) -> Dict:
people_events_table = metadata.tables["people_events"]
people_table = metadata.tables["people"]
result = connection.execute(
people_table.select().join(
people_events_table,
people_table.c.id == people_events_table.c.event_id
).where(
(people_events_table.c.people_id == people_id)
& (people_events_table.c.event_id == event_id)
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
@app.route("/people/<int:people_id>/events/<int:event_id>", methods=["PATCH"])
def patch_people_events(people_id: int, event_id: int) -> Response:
# Ideally there'd be error handling for if the event doesn't exist here
# Sign person up for event
signup_user(people_id, event_id)
return Response(status=204) # No Content
@app.route("/events/<int:event_id>/people/<int:people_id>", methods=["PATCH"])
def patch_events_people(event_id: int, people_id: int) -> Response:
# Ideally there'd be error handling for if the event doesn't exist here
# Sign person up for event
signup_user(people_id, event_id)
return Response(status=204) # No Content
@app.route("/people/<int:people_id>/events/<int:event_id>", methods=["DELETE"])
def delete_people_events(people_id: int, event_id: int) -> Response:
delete_signup(people_id, event_id)
return Response(status=204) # No Content
@app.route("/events/<int:event_id>/people/<int:people_id>", methods=["DELETE"])
def delete_events_people(event_id: int, people_id: int) -> Response:
delete_signup(people_id, event_id)
return Response(status=204) # No Content
if __name__ == "__main__":
app.run()Code language: Python (python)One to One relationships don’t need sub-endpoints
The same concept can also be applied for 1-1 relationships, but when there’s only ever one relationship it doesn’t make as much sense to make their own endpoints, so instead a ‘links’ field is often added to a resource.
Meaning that the return of /events/<int: events_id> changes from
{
"result":[
{
"date":"Thu, 30 Jun 2022 00:00:00 GMT",
"id":2,
"type":"conference"
}
]
}Code language: JSON / JSON with Comments (json)to
{
"result":[
{
"date":"Sat, 20 Dec 2025 00:00:00 GMT",
"id":3,
"links":{
"bill":"/bills/3"
},
"type":"holiday party"
}
]
}Code language: JSON / JSON with Comments (json)from typing import Dict
import sqlalchemy
from sqlalchemy.orm import sessionmaker
from flask import Flask
app = Flask(__name__)
# Connect to the DB and reflect metadata.
engine = sqlalchemy.create_engine("postgresql://coderpad:@/coderpad?host=/tmp/postgresql/socket")
connection = engine.connect()
Session = sessionmaker(bind=engine)
session = Session()
metadata = sqlalchemy.MetaData()
metadata.reflect(bind=engine)
@app.route("/events/<int:id>", methods=["GET"])
def get_one_events_with_links(id: int) -> Dict:
# Ideally you'd simplify this so you aren't making so many Database calls,
# but it's left this way for clarity.
events_table = metadata.tables["events"]
bills_table = metadata.tables["bills"]
result = connection.execute(
sqlalchemy.select(events_table).where(
events_table.c.id == id
)
).fetchall()
events = []
for event in result:
event = dict(event)
# Get related bill
bill = connection.execute(
sqlalchemy.select(bills_table).where(
bills_table.c.event_id == id
)
).fetchall()
if len(bill) > 0:
bill_id = dict(bill[0]).get("id")
event["links"] = {
"bill": "/bills/{}".format(bill_id)
}
events.append(event)
return {
"result": events
}
if __name__ == "__main__":
app.run()Code language: Python (python)One to Many relationships combine One to One and Many to Many approaches
For one to many relationships, such as how one person can have multiple bills, but one bill only has one person who has to pay it, we combine both concepts. From the side of a resource with many related concepts (people) we reuse the same concept as many-to-many, leaving us with the endpoints:
/people/<int: people_id>/bills/ with GET
/people/<int: people_id>/bills/<int: bill_id> with GET, PATCH, and DELETECode language: HTML, XML (xml)However, “PATCH” and “DELETE” are also redundant since any editing of ‘bills‘ can be done on the /bills/<int: bill_id> endpoints, so we can ignore those verbs here.
To finish off the people to bills relationship, we reuse the 1-1 concept of a `links` field for the other side of the equation, meaning that the return of /bills/<int: id> changes from
{
"result":[
{
"cost":10000,
"due":"Tue, 01 Jun 2021 00:00:00 GMT",
"event_id":2,
"id":2,
"paid":"not paid",
"payer_id":1
}
]
}Code language: JSON / JSON with Comments (json)to
{
"result":[
{
"cost":10000,
"due":"Tue, 01 Jun 2021 00:00:00 GMT",
"event_id":2,
"id":2,
"links":{
"people":"/people/1"
},
"paid":"not paid",
"payer_id":1
}
]
}Code language: JSON / JSON with Comments (json)from typing import Dict
import sqlalchemy
from sqlalchemy.orm import sessionmaker
from flask import Flask
app = Flask(__name__)
# Connect to the DB and reflect metadata.
engine = sqlalchemy.create_engine("postgresql://coderpad:@/coderpad?host=/tmp/postgresql/socket")
connection = engine.connect()
Session = sessionmaker(bind=engine)
session = Session()
metadata = sqlalchemy.MetaData()
metadata.reflect(bind=engine)
@app.route("/events/<int:id>", methods=["GET"])
def get_one_events_with_links(id: int) -> Dict:
# Ideally you'd simplify this so you aren't making so many Database calls,
# but it's left this way for clarity.
events_table = metadata.tables["events"]
bills_table = metadata.tables["bills"]
result = connection.execute(
sqlalchemy.select(events_table).where(
events_table.c.id == id
)
).fetchall()
events = []
for event in result:
event = dict(event)
# Get related bill
bill = connection.execute(
sqlalchemy.select(bills_table).where(
bills_table.c.event_id == id
)
).fetchall()
if len(bill) > 0:
bill_id = dict(bill[0]).get("id")
event["links"] = {
"bill": "/bills/{}".format(bill_id)
}
events.append(event)
return {
"result": events
}
@app.route("/bills/<int:id>", methods=["GET"])
def get_one_bills(id: int) -> Dict:
people_table = metadata.tables["people"]
bills_table = metadata.tables["bills"]
result = connection.execute(
sqlalchemy.select(bills_table).where(
bills_table.c.id == id
)
).fetchall()
bills = []
for bill in result:
bill = dict(bill)
# Get related person
person = connection.execute(
sqlalchemy.select(people_table).where(
people_table.c.id == bill.get("payer_id")
)
).fetchall()
if len(person) > 0:
people_id = dict(person[0]).get("id")
bill["links"] = {
"people": "/people/{}".format(people_id)
}
bills.append(bill)
return {
"result": bills
}
if __name__ == "__main__":
app.run()Code language: Python (python)All API Endpoints
To summarize, for our four database tables (people, events, bills, and people_events, our API looks like:
/people with GET
/people/<int:id> with GET
/people with POST
/people/<int: id> with PATCH
/people/<int: id> with DELETE
/people/<int: people_id>/events/ with GET
/people/<int: people_id>/events/<int: event_id> with GET, PATCH, and DELETE
/people/<int: people_id>/bills/ with GET
/people/<int: people_id>/bills/<int: bill_id> with GET
/events with GET
/events/<int:id> with GET
/events with POST
/events/<int: id> with PATCH
/events/<int: id> with DELETE
/events/<int: event_id>/people/ with GET
/events/<int: event_id>/people/<int: people_id> with GET, PATCH, and DELETE
/bills with GET
/bills/<int:id> with GET
/bills with POST
/bills/<int: id> with PATCH
/bills/<int: id> with DELETECode language: HTML, XML (xml)And the code for it all looks like:
import sqlalchemy
from sqlalchemy.orm import sessionmaker
from flask import Flask, Response, request
from typing import Any, Dict
app = Flask(__name__)
# Connect to the DB and reflect metadata.
engine = sqlalchemy.create_engine("postgresql://coderpad:@/coderpad?host=/tmp/postgresql/socket")
connection = engine.connect()
Session = sessionmaker(bind=engine)
session = Session()
metadata = sqlalchemy.MetaData()
metadata.reflect(bind=engine)
def get_all(table_name: str) -> Dict:
table = metadata.tables[table_name]
result = connection.execute(table.select()).fetchall()
return {
"result": [
dict(resource) for resource in result
]
}
def get_one(table_name: str, column_name: str, value: Any) -> Dict:
table = metadata.tables[table_name]
result = connection.execute(
sqlalchemy.select(table).where(
getattr(table.c, column_name) == value
)
).fetchall()
return {
"result": [
dict(resource) for resource in result
]
}
def patch_one(table_name: str, column_name: str, value: Any) -> Response:
content_type = request.headers.get("Content-Type")
if (content_type != "application/json"):
return {
"result": "Failure",
"reason": "Content-Type not supported!",
}
table = metadata.tables[table_name]
# There should be validation that the PATCH body is valid here!
connection.execute(
sqlalchemy.update(table).where(
getattr(table.c, column_name) == value
).values(
**request.json,
)
)
return Response(status=204) # No Content
def post(table_name: str) -> Response:
content_type = request.headers.get("Content-Type")
if (content_type != "application/json"):
return {
"result": "Failure",
"reason": "Content-Type not supported!",
}
table = metadata.tables[table_name]
# There should be validation that the POST body is valid here!
connection.execute(
sqlalchemy.insert(table).values(
**request.json,
)
)
return Response(status=201) # Created
def delete_one(table_name: str, column_name: str, value: Any) -> Response:
table = metadata.tables[table_name]
connection.execute(
sqlalchemy.delete(table).where(
getattr(table.c, column_name) == value
)
)
return Response(status=204) # No Content
def signup_user(people_id: int, event_id: int) -> None:
people_events_table = metadata.tables["people_events"]
connection.execute(
sqlalchemy.insert(people_events_table).values(
people_id=people_id,
event_id=event_id,
)
)
def delete_signup(people_id: int, event_id: int) -> None:
people_events_table = metadata.tables["people_events"]
connection.execute(
sqlalchemy.delete(people_events_table).where(
(people_events_table.c.people_id == people_id)
& (people_events_table.c.event_id == event_id)
)
)
# Endpoints:
@app.route("/people", methods=["GET"])
def get_all_people() -> Dict:
return get_all(table_name="people")
@app.route("/people/<int:id>", methods=["GET"])
def get_one_people(id: int) -> Dict:
return get_one(
table_name="people",
column_name="id",
value=id,
)
@app.route("/people", methods=["POST"])
def post_people() -> Response:
return post("people")
@app.route("/people/<int:id>", methods=["PATCH"])
def patch_people(id: int) -> Response:
return patch_one(
table_name="people",
column_name="id",
value=id,
)
@app.route("/people/<int:id>", methods=["DELETE"])
def delete_people(id: int) -> Response:
# Should cascade deleting relationship(s) in `people_events` here
# Should also cascade deleting related `bill`(s) here
return delete_one(
table_name="people",
column_name="id",
value=id,
)
# Without 'links'
# @app.route("/events", methods=["GET"])
# def get_all_events() -> Dict:
# return get_all(table_name="events")
@app.route("/events", methods=["GET"])
def get_all_events() -> Dict:
# Ideally you'd simplify this so you aren't making so many Database calls,
# but it's left this way for clarity.
events_table = metadata.tables["events"]
bills_table = metadata.tables["bills"]
result = connection.execute(events_table.select()).fetchall()
events = []
for event in result:
event = dict(event)
# Get related bill
bill = connection.execute(
sqlalchemy.select(bills_table).where(
bills_table.c.event_id == event.get("id")
)
).fetchall()
if len(bill) > 0:
bill_id = dict(bill[0]).get("id")
event["links"] = {
"bill": "/bills/{}".format(bill_id)
}
events.append(event)
return {
"result": events
}
# Without 'links'
# @app.route("/events/<int:id>", methods=["GET"])
# def get_one_events(id: int) -> Dict:
# return get_one(
# table_name="events",
# column_name="id",
# value=id,
# )
@app.route("/events/<int:id>", methods=["GET"])
def get_one_events_with_links(id: int) -> Dict:
# Ideally you'd simplify this so you aren't making so many Database calls,
# but it's left this way for clarity.
events_table = metadata.tables["events"]
bills_table = metadata.tables["bills"]
result = connection.execute(
sqlalchemy.select(events_table).where(
events_table.c.id == id
)
).fetchall()
events = []
for event in result:
event = dict(event)
# Get related bill
bill = connection.execute(
sqlalchemy.select(bills_table).where(
bills_table.c.event_id == id
)
).fetchall()
if len(bill) > 0:
bill_id = dict(bill[0]).get("id")
event["links"] = {
"bill": "/bills/{}".format(bill_id)
}
events.append(event)
return {
"result": events
}
@app.route("/events", methods=["POST"])
def post_events() -> Response:
return post("events")
@app.route("/events/<int:id>", methods=["PATCH"])
def patch_events(id: int) -> Response:
return patch_one(
table_name="events",
column_name="id",
value=id,
)
@app.route("/events/<int:id>", methods=["DELETE"])
def delete_events(id: int) -> Response:
# Should cascade deleting relationship(s) in `people_events` here
# Should also cascade deleting related `bill` here
return delete_one(
table_name="events",
column_name="id",
value=id,
)
# Without 'links'
# @app.route("/bills", methods=["GET"])
# def get_all_bills() -> Dict:
# return get_all(table_name="bills")
@app.route("/bills", methods=["GET"])
def get_all_bills() -> Dict:
people_table = metadata.tables["people"]
bills_table = metadata.tables["bills"]
result = connection.execute(bills_table.select()).fetchall()
bills = []
for bill in result:
bill = dict(bill)
# Get related person
person = connection.execute(
sqlalchemy.select(people_table).where(
people_table.c.id == bill.get("payer_id")
)
).fetchall()
if len(person) > 0:
people_id = dict(person[0]).get("id")
bill["links"] = {
"people": "/people/{}".format(people_id)
}
bills.append(bill)
return {
"result": bills
}
# Without 'links'
# @app.route("/bills/<int:id>", methods=["GET"])
# def get_one_bills(id: int) -> Dict:
# return get_one(
# table_name="bills",
# column_name="id",
# value=id,
# )
@app.route("/bills/<int:id>", methods=["GET"])
def get_one_bills(id: int) -> Dict:
people_table = metadata.tables["people"]
bills_table = metadata.tables["bills"]
result = connection.execute(
sqlalchemy.select(bills_table).where(
bills_table.c.id == id
)
).fetchall()
bills = []
for bill in result:
bill = dict(bill)
# Get related person
person = connection.execute(
sqlalchemy.select(people_table).where(
people_table.c.id == bill.get("payer_id")
)
).fetchall()
if len(person) > 0:
people_id = dict(person[0]).get("id")
bill["links"] = {
"people": "/people/{}".format(people_id)
}
bills.append(bill)
return {
"result": bills
}
@app.route("/bills", methods=["POST"])
def post_bills() -> Response:
return post("bills")
@app.route("/bills/<int:id>", methods=["PATCH"])
def patch_bills(id: int) -> Response:
return patch_one(
table_name="bills",
column_name="id",
value=id,
)
@app.route("/bills/<int:id>", methods=["DELETE"])
def delete_bills(id: int) -> Response:
return delete_one(
table_name="bills",
column_name="id",
value=id,
)
# Many to Many relationships:
@app.route("/people/<int:people_id>/events", methods=["GET"])
def get_people_events_all(people_id: int) -> Dict:
people_events_table = metadata.tables["people_events"]
events_table = metadata.tables["events"]
result = connection.execute(
events_table.select().join(
people_events_table,
events_table.c.id == people_events_table.c.event_id
).where(
people_events_table.c.people_id == people_id
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
@app.route("/people/<int:people_id>/events/<int:event_id>", methods=["GET"])
def get_people_events_one(people_id: int, event_id: int) -> Dict:
people_events_table = metadata.tables["people_events"]
events_table = metadata.tables["events"]
result = connection.execute(
events_table.select().join(
people_events_table,
events_table.c.id == people_events_table.c.event_id
).where(
(people_events_table.c.people_id == people_id)
& (people_events_table.c.event_id == event_id)
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
@app.route("/events/<int:event_id>/people", methods=["GET"])
def get_events_people_all(event_id: int) -> Dict:
people_events_table = metadata.tables["people_events"]
people_table = metadata.tables["people"]
result = connection.execute(
people_table.select().join(
people_events_table,
people_table.c.id == people_events_table.c.event_id
).where(
people_events_table.c.event_id == event_id
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
@app.route("/events/<int:event_id>/people/<int:people_id>", methods=["GET"])
def get_events_people_one(event_id: int, people_id: int) -> Dict:
people_events_table = metadata.tables["people_events"]
people_table = metadata.tables["people"]
result = connection.execute(
people_table.select().join(
people_events_table,
people_table.c.id == people_events_table.c.event_id
).where(
(people_events_table.c.people_id == people_id)
& (people_events_table.c.event_id == event_id)
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
@app.route("/people/<int:people_id>/events/<int:event_id>", methods=["PATCH"])
def patch_people_events(people_id: int, event_id: int) -> Response:
# Ideally there'd be error handling for if the event doesn't exist here
# Sign person up for event
signup_user(people_id, event_id)
return Response(status=204) # No Content
@app.route("/events/<int:event_id>/people/<int:people_id>", methods=["PATCH"])
def patch_events_people(event_id: int, people_id: int) -> Response:
# Ideally there'd be error handling for if the event doesn't exist here
# Sign person up for event
signup_user(people_id, event_id)
return Response(status=204) # No Content
@app.route("/people/<int:people_id>/events/<int:event_id>", methods=["DELETE"])
def delete_people_events(people_id: int, event_id: int) -> Response:
delete_signup(people_id, event_id)
return Response(status=204) # No Content
@app.route("/events/<int:event_id>/people/<int:people_id>", methods=["DELETE"])
def delete_events_people(event_id: int, people_id: int) -> Response:
delete_signup(people_id, event_id)
return Response(status=204) # No Content
# 1 to Many Relationships:
@app.route("/people/<int:people_id>/bills", methods=["GET"])
def get_people_bills_all(people_id: int) -> Dict:
bills_table = metadata.tables["bills"]
people_table = metadata.tables["people"]
result = connection.execute(
bills_table.select().join(
people_table,
people_table.c.id == bills_table.c.payer_id
).where(
(people_table.c.id == people_id)
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
@app.route("/people/<int:people_id>/bills/<int:bill_id>", methods=["GET"])
def get_people_bills_one(people_id: int, bill_id: int) -> Dict:
bills_table = metadata.tables["bills"]
people_table = metadata.tables["people"]
result = connection.execute(
bills_table.select().join(
people_table,
people_table.c.id == bills_table.c.payer_id
).where(
(bills_table.c.id == bill_id)
& (people_table.c.id == people_id)
)
).fetchall()
return {
"result": [
dict(event) for event in result
]
}
if __name__ == "__main__":
app.run()Code language: Python (python)This gives the user full CRUD functionality over all pieces of data and their relationships.
We could always make the many to many relationships nicer to use for a human (by doing things such as returning all pieces of data in one endpoint), and there are styles of APIs that do this, but it’s worth remembering that APIs are meant for machines, so the easiest way to programmatically use an API is the priority when designing them.
In the end, what all of this means is that we use relationships in SQL to minimize data duplication, and we use the four REST verbs in an API to allow code written by others to easily interact with our data.
The implementation of said API involves things like inline for loops to build lists which allows for shorter, easier to read code. This concept is called “list comprehension”, and if you’d like to read more about it, CoderPad has a guide on how to write them yourself.
Similarly, if you’re looking to expand your knowledge of Python’s internals, maybe take a look at CoderPad’s guide to Python’s Magic Methods, which can be used for things like getting attributes of data.
These concepts can also be applied to other types of databases, so if you’re interested it’d be worth it to check out CoderPad’s guide to SQL versus NoSQL.
Jennifer is a full stack developer with a passion for all areas of software development. He loves being a polyglot of programming languages and teaching others what he’s learned.