How to interact with a program
API is any interface that allows us to program an application
Talking to the filesystem - Filesystem API
Talking to git
- git API
Talking to a webserver - web API
The focus here is on web APIs - how to exchange data with a webservice
HTTP (Hypertext Transfer Protocol) is the backbone of the internet. Every time you visit a webpage, you use HTTP
In HTTP lingo, a URL (Uniform Resource Locator) points to a given resource.
For example: www.github.com/andersbogsnes
has two parts
The host or name of the server www.github.com
and the resource /andersbogsnes
A resource has a representation on the server - it could be a file, such as an HTML page, or it could be information stored in a database.
The requested resource is sent back to the client in a representation - usually some form of text.
HTTP is a protocol that defines how messages are passed between a client - your machine - and the webserver.
For example, the message for requesting Google’s homepage looks like this:
GET / HTTP/1.1
Host: www.google.com
GET # Method / # Resource HTTP/1.1 # Protocol
GET
the /
resource from google.We will focus on the methods that get used most commonly:
The GET request simply requests a resource. Should only ever retrieve data
A GET that only returns the header and not the body. Commonly used for health checks
Submit a resource to the server. Usually used to change state on the server, e.g create a new resource
Submit a resoruce intended to replace an existing resource.
Delete a given resource from the server
Modify part of an existing resource
Google’s webserver will respond to our Request with a Response message
HTTP/1.1 200 OK
<html>...</html>
In the Response, the header contains the protocol and the status code
HTTP/1.1 # HTTP Protocol 200 OK # Status Code
Status codes are the server’s main method of communicating the result of the Request.
There are many status codes with lots of nuance, but a few you might know:
Most people will only use HTTP when they use a browser to navigate to a webpage
When you navigate to google.com, your browser does a GET for the website.
The Response body is the text of the website HTML which your browser knows how to render.
You might want to create a new user - the form you fill out gets POSTed to google.com, which creates a new resource - your new user account
An API is designed to use the same methods to manipulate data.
Take the Twitter API - a tweet can be thought of as a resource!
Let’s write some code - install fastapi + dependencies in a new virtualenv
pip install fastapi[all]
from fastapi import FastAPI
api = FastAPI()
@api.get("/")
def hello_world():
return "Hello world"
>>> uvicorn --reload hello_world:api
INFO: Started server process [18657]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
Note how the API is defined
A method + route is mapped to a function that gets executed when the Server receives a request for that combination
Navigate to http://127.0.0.1:8000
Rememember, your browser will create a GET HTTP message for /
The request triggers the mapped function to generate a reponse
FastAPI handles calling the function and converting the return value to something the browser can understand
We can create a dynamic endpoint by creating a route placeholder
@api.get("/hello/{name}")
def hello_world_name(name: str):
return f"Hello world {name}!"
Try navigating to http://127.0.0.1:8000/hello/foo
The mapped endpoint now takes a named argument, which FastAPI will give us as a variable and we can use it in our function.
⚠️
FastAPI matches by name, so the placeholder name and the function argument name must be the same
@api.get("/hello/{name}/{greeting}")
def hello_world_name_and_greeting(name: str, greeting: str):
return f"{greeting} {name}"
One of the features of FastAPI is using the pydantic
library to do data validation.
Pydantic is a FastAPI dependency, and is used to do data validation using python typehinting
Let’s define a User schema
from pydantic import BaseModel
class UserSchema(BaseModel):
username: str
email: str
We can now use that in a post
request handler
@api.post("/user")
def create_user(user: UserSchema):
return "Thank you new user"
How do we make a POST request?
Navigate to 127.0.0.1:8000/docs
FastAPI automatically creates an OpenAPI spec for the API
Find the endpoint and Try it out
Notice that any docstrings you include in your handler function gets included here
On the commandline we can use curl
curl -X POST localhost:8000/user -d '{"username": "Anders", "email": "andersbogsnes@gmail.com"}'
Pycharm has a http request feature
Right-click -> New File -> HTTP Request.
The syntax for a POST looks like this:
POST http://localhost:8000/user
Content-Type: application/json
{"username": "anders", "email": "andersbogsnes@gmail.com"}
Choose one and try it!
Note the data payload used here is JSON format
The lingua franca of APIs these days is JSON - JavaScript Object Notation
JSON is written using two elements
The example data is a key-> value mapping
{"username": "Anders", "email": "andersbogsnes@gmail.com"}
It could also be an array of multiple users:
[
{"username": "Anders", "email": "andersbogsnes@gmail.com"},
{"username": "Tom", "email": "tomhanks@gmail.com"}
]
We can return a dictionary and FastAPI will convert it to JSON and return it
@api.post("/user")
def create_new(user: UserSchema):
# This is a python dictionary, not JSON!
return {
"name": user.username,
"email": user.email
}
Since we are using pydantic, we can see what data is available on the object!
We can also define a response model to validate the outgoing data - maybe we only want the email to be returned?
class UserOutSchema(BaseModel):
email: str
@api.post("/user", response_model=UserOutSchema)
def create_new_user(user: UserSchema):
return {
# What happens if the key is 'name'?
"username": user.username,
"email": user.email
}
What is the response from the API now?
The typehints we’ve used in pydantic helps to validate the data, but str
is generic - anything can be a string.
Pydantic comes with built-in validators we can use. Let’s validate that the email is actually an email
Change the UserSchema to be of type EmailStr instead
from pydantic import EmailStr, BaseModel
class UserSchema(BaseModel):
username: str
email: EmailStr
What happens if you try to post a non-email?
Let’s build an AirBnB-alike API.
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
class ListingOutSchema(BaseModel):
username: str
address: str
price: float
class ListingSchema(ListingOutSchema):
email: EmailStr
api = FastAPI()
LISTINGS = [{
"email": "test@test.com",
"username": "test",
"address": "Nørrebrogade 20",
"price": 700
}]
@api.post("/listing", response_model=ListingOutSchema)
def create_new_listing(listing: ListingSchema):
return listing
@api.get("/listing/{listing_id}", response_model=ListingOutSchema)
def get_listing(listing_id: int):
return LISTINGS[listing_id]
A core component of an API is data
We want to pass data back and forth to our API using JSON
Data needs to be stored - generally we want a database!
When working with APIs, we generally want to fetch a row at a time or a limited queryset
The SQLAlchemy ORM (Object Relational Mapper) is great for working with rows of data
An ORM represents a row in the database as a Python object and makes it easy to manipulate it
First, we need to install sqlalchemy:
pip install --pre sqlalchemy # For sqlalchemy==1.4
⚠️ SQLAlchemy 2.0 is around the corner. We want to use 1.4
which has support for 2.0
functionality, but it’s still beta
SQLAlchemy lets us declare a Table
class that represents a row in our database. To do this, SQLAlchemy needs to generate a Base
class to inherit from
from sqlalchemy.orm import declarative_base
Base = declarative_base()
Now we can create our mapping class
import sqlalchemy as sa
class Listing(Base):
__tablename__ = "listings"
id = sa.Column(sa.Integer, primary_key=True)
username = sa.Column(sa.String)
email = sa.Column(sa.String)
address = sa.Column(sa.String)
price = sa.Column(sa.Float)
We don’t need to provide an __init__
method - SQLAlchemy autogenerates one.
You can, however, add an __init__
if you prefer
SQLAlchemy will also autogenerate the id
column when we add a new row
We have a mapping, but no database yet - we can ask SQLAlchemy to write the SQL needed to match our declaration
# Set the future flag to opt-in to 2.0 behaviour
engine = sa.create_engine("sqlite:///airbnb.db", future=True)
# All tables inheriting from Base
# are registered in the metadata object
Base.metadata.create_all(engine) # Execute the sql with the engine
We can now insert some data into the database using our ORM class
# Make an instance of Listing
listing_1 = Listing(username="Anders",
email="andersbogsnes@gmail.com",
address="Nørrebrogade 20",
price=700)
# The context manager automatically closes the session
with Session(engine) as session:
session.add(listing_1)
session.commit()
To read the data, we must create a SELECT statement and execute it in a Session
sql = sa.select(Listing).where(Listing.username == "Anders")
with Session(engine) as session:
# Get the first result
result = session.execute(sql).first()
>>> print(result.Listing.username)
"Anders"
The result we get back from sqlalchemy is a Row
object with columns for each item in the select
.
When using the ORM, we generally just want the object, which we can get with .scalars
with Session(engine) as session:
result = session.execute(sql).scalars().first()
>>> print(result.username)
"Anders"
When working with inserts, there are two principles we want to follow
The Session creates a database connection, which needs to be closed.
Lots of opens connections is bad!
But having to open a new connection constantly is also bad...
We want to align our session being open with the duration of the request
When we commit, we hit “save” on the database and data is transferred and written.
Too often and we transfer unnecessary data
Too little and we risk losing changes
Let’s get started on the CRUD (Create/Read/Update/Delete) logic for our api
def create_new_listing(session: Session,
username: str,
email: str,
address: str,
price: float
) -> Listing:
listing = Listing(username=username,
email=email,
address=address,
price=price)
session.add(listing)
return listing
def get_listing_by_id(session: Session, listing_id: int) -> Listing:
# This uses the special `get` method to lookup by primary key
result = session.get(Listing, listing_id)
# Alternative syntax
# stmt = sa.select(Listing).where(Listing.id == listing_id)
# result = session.execute(stmt).scalar()
return result
To update a record, we can simply modify the attribute and add it back to the session
with Session(engine) as session:
listing = get_listing(session, 1)
listing.username = "Anders Bogsnes"
session.add(listing)
session.commit()
To delete is also simple
with Session(engine) as session:
listing = get_listing(session, 1)
session.delete(listing)
session.commit()
One of the nice features of ORM is that we can easily setup relationships between tables, and then any related tables will be available as an attribute on the Table class
We won’t be using this feature in our very simple API, but it’s good to know about
A better design for our database might be to have a Users
table and a Listings
table and be able to join the two.
It would be nice to be able to access all listings for a given user from a user
instance:
user = get_user(user_id=1)
listings = user.listings
SQLALchemy will implement this for us if we specify a relationship
from sqlalchemy.orm import relationship
class User(Base):
__tablename__ = "user"
id = Column(Integer, primary_key=True)
name = Column(String)
listings = relationship("Listing", back_populates="user")
class Listing(Base):
id = Column(Integer, primary_key=True)
user_id = Column(Integer, sa.ForeignKey('user.id'))
address = Column(String)
user = relationship("User", back_populates="listings")
Given this declaration, SQLAlchemy will handle the SQL for looking up the associated rows when we do user.listings
.
As we have declared ForeignKeys, SQLAlchemy will automatically use those for the join
Now we have the basic CRUD setup, we need to hook up our handler functions to use the database
Remember, we want our connections to be open for the duration of the request and then close it.
FastAPI lets us Depend
on a function to ensure that it will run once per request-response cycle
First we define a get_session
function
def get_session():
with Session(engine) as session:
yield session
Change the endpoint
@api.get("/listing/{listing_id}", response_model=ListingOutSchema)
def get_listing(listing_id: int, db: Session = Depends(get_db)):
return get_listing_by_id(session=db, listing_id=listing_id)
What happens when we run this?
Pydantic expects to receive dictionaries, and we are passing it a Listing
object.
Luckily, Pydantic objects can be put into ORM mode
to allow passing objects, like a Listing
class ListingOutSchema(BaseModel):
username: str
address: str
price: float
class Config:
orm_mode = True
Turn the API into a full CRUD application for handling listings. Think about what HTTP methods to use for each one
BONUS POINTS: Try to write a route that gets all listings
from typing import List
import sqlalchemy as sa
from fastapi import FastAPI, Depends, status
from pydantic import BaseModel, EmailStr
from sqlalchemy.orm import declarative_base, Session
Base = declarative_base()
engine = sa.create_engine("sqlite:///airbnb.db", future=True)
api = FastAPI()
class Listing(Base):
__tablename__ = "listings"
id = sa.Column(sa.Integer, primary_key=True)
username = sa.Column(sa.String)
email = sa.Column(sa.String)
address = sa.Column(sa.String)
price = sa.Column(sa.Float)
class ListingOutSchema(BaseModel):
username: str
address: str
price: float
class Config:
orm_mode = True
class ListingInSchema(ListingOutSchema):
email: EmailStr
def create_new_listing(session: Session,
username: str,
email: str,
address: str,
price: float
) -> Listing:
listing = Listing(username=username,
email=email,
address=address,
price=price)
session.add(listing)
return listing
def get_listing_by_id(session: Session, listing_id: int) -> Listing:
return session.get(Listing, listing_id)
def delete_listing_by_id(session: Session, listing_id: int) -> None:
listing = get_listing_by_id(session, listing_id)
session.delete(listing)
def update_listing_by_id(session: Session,
listing_id: int,
data: ListingInSchema
) -> Listing:
listing = get_listing_by_id(session, listing_id)
for key, val in data.dict().items():
if hasattr(listing, key):
setattr(listing, key, val)
session.add(listing)
return listing
def get_all_listings_from_db(session: Session) -> List[Listing]:
stmt = sa.select(Listing)
return session.execute(stmt).scalars().all()
def get_db():
with Session(engine) as session:
yield session
@api.get("/listing/{listing_id}",
response_model=ListingOutSchema)
def get_listing(listing_id: int, db: Session = Depends(get_db)):
return get_listing_by_id(session=db,
listing_id=listing_id)
@api.post("/listing",
response_model=ListingOutSchema,
status_code=status.HTTP_201_CREATED)
def create_listing(listing: ListingInSchema,
db: Session = Depends(get_db)):
new_listing = create_new_listing(session=db,
**listing.dict())
db.commit()
return new_listing
@api.delete("/listing/{listing_id}",
status_code=status.HTTP_204_NO_CONTENT)
def delete_listing(listing_id: int,
db: Session = Depends(get_db)):
delete_listing_by_id(session=db, listing_id=listing_id)
db.commit()
@api.patch("/listing/{listing_id}",
response_model=ListingOutSchema)
def update_listing(listing_id: int,
listing: ListingInSchema,
db: Session = Depends(get_db)):
updated_listing = update_listing_by_id(session=db,
listing_id=listing_id,
data=listing)
db.commit()
return updated_listing
@api.get("/listings",
response_model=List[ListingOutSchema])
def get_all_listings(db: Session = Depends(get_db)):
return get_all_listings_from_db(db)
In the Requests-Response cycle, all the responses currently look like this
HTTP/1.1 200 OK
{some data}
By default, FastAPI handles sets these status codes automatically
We should be more granular in status codes, but we have to tell FastAPI what is appropriate in each case
Rest API Tutorials has an overview of suggestions
For convenience, FastAPI has defined status codes as constants to make the code more readable, you can pass an int directly if you prefer
from fastapi import status
@api.post("/listing",
response_model=ListingOutSchema,
status_code=status.HTTP_201_CREATED)
def create_new_listing(listing: ListingInSchema,
db: Session = Depends(get_db)):
return create_new_listing(session=db,
**listing.dict())
Some common status codes:
There are some other ways a client can pass information to the server and one important one is the Header.
Remember the HTTP Request Message - it actually looks something like this
GET / HTTP/1.1
Host: github.com
User-Agent: Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:83.0) Gecko/20100101 Firefox/83.0
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8
Accept-Language: en-US,en;q=0.5
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
Upgrade-Insecure-Requests: 1
Pragma: no-cache
Cache-Control: no-cache
The rest of the message is passing Headers, which the server can interpret and modify it’s behaviour accordingly.
Headers are simple key-value pairs passed in HTTP messages and can be found in both response and request messages
In FastAPI, it’s simple to get header values
from fastapi import Header
@api.get("/user-agent")
def get_user_agent(user_agent: str = Header(default=None)):
return f"Your user agent is {user_agent}"
The most common header we have to worry about is authentication. A typical usecase is Bearer Tokens
The client can send its authentication as a header with the format Authorization: Bearer mytoken
The second most common way of passing parameters is via query params
You might have noticed this in various queries you’ve made on the internet.
Let’s look at DAWA’s Autocomplete API (Danmarks Adresser Web API)
To autocomplete an address in zipcode 2720 starting with Bog
we have to use the following URL
?
marks the beginning of the query parametersq=Bog
is now a query parameter - the text we are searching for according to the docs&
marks a new query parameterpostnr=2720
passes 2720 to the query parameter postnr
The documentation notes several other query parameters we could have passed
In FastAPI, it’s very simple to get query params.
If we wanted to recreate the DAWA API
@api.get("/autocomplete")
def find_suggestions(q: str, postnr: Optional[int] = None):
...
FastAPI will automatically parse the query parameters for you. It can also figure out what parts are path parameters and which ones are query parameters
Testing is one of the most important parts of building an API and FastAPI makes it simple
# The FastAPI instance
from app import app
from fastapi.testclient import TestClient
client = TestClient(app)
def test_get_docs():
response = client.get("/docs")
assert response.status_code == 200
assert "FastAPI" in response.text
The TestClient wraps the app in a requests.Session
, so that we can use the requests
API directly to talk to our app
When structuring the app, much like any software project, it’s important to think about testing.
When building APIs, think about splitting up the functionality into separate components
This structure allows us to test routes separately from business logic and allows for easy patching when testing routes
class TestCanPostListing:
@pytest.fixture(scope="class")
def listing(self):
return Listing(username="test",
email="test@test.com",
address="Test Adress",
price=700)
@pytest.fixture(scope="class")
def patched_service(self,
class_mocker: MockFixture,
listing: Listing) -> MagicMock:
return class_mocker.patch("airbnb.routes.services.create_new_listing",
return_value=listing)
@pytest.fixture(scope="class")
def response(self,
client: TestClient,
listing: Listing,
patched_service: MagicMock):
return client.post("/listing",
json={"username": listing.username,
"email": listing.email,
"address": listing.address,
"price": listing.price})
def test_response_has_correct_status_code(self, response: Response):
assert response.status_code == 201
def test_response_has_correct_data(self,
listing: Listing,
response: Response):
expected = {
"username": listing.username,
"address": listing.address,
"price": listing.price
}
assert response.json() == expected
import pytest
from sqlalchemy import select
from sqlalchemy.orm import Session
from airbnb.models import Listing
from airbnb.services import create_new_listing
class TestCanCreateListing:
@pytest.fixture(scope="class")
def listing(self) -> Listing:
return Listing(username="test",
email="test@test.com",
address="Test Address",
price=700)
@pytest.fixture()
def new_listing(self, listing: Listing, db: Session):
return create_new_listing(session=db,
email=listing.email,
username=listing.username,
address=listing.address,
price=listing.price)
def test_can_create_new_listing(self,
new_listing: Listing,
listing: Listing):
assert new_listing == listing
def test_insert_listing_twice_creates_two_rows(self,
listing: Listing,
db: Session):
for _ in range(2):
create_new_listing(db,
username=listing.username,
email=listing.email,
address=listing.address,
price=listing.price
)
results = db.execute(select(Listing)).scalars().all()
assert len(results) == 2
When designing something, the first thing to think about is naming
-
/user/{user_id}/address
Think about what actions the user can do
Write them down and make sure you have an endpoint that let’s them do that
The API is getting unsanitized input from the internet - never trust the internet!
Assume data is wrong and handle it - your server should never crash from bad data
BOTUS
? What should the response be?In the context of “Talking to Webservers”, there are many methods to do that.
The most common among modern APIs is REST, REpresentational State Transfer
The term REST was first defined in 2000 to describe a way of modelling data using HTTP
The original tenets of REST were:
Allowing the client - the consumer of data - to be separate from the server - the producer of data. Also known as frontend + backend
One of the most important principles - there is no state stored. Any API call is completely self-sufficient and does not rely on anything else
Responses from the API should be able to be cached
The API should be able to handle any number of intermediaries between itself and the client. E.g. load balancers, proxies.
Probably the least common, but the API should be able to extend server behaviour, by accepting executable code to run server-side
The REST spec identifies a Uniform Interface to be as follows:
The request identifies the resource the client needs, but the representation of that resource is not constrained by the server’s internal representation. For example, the server could send data from its database as HTML, XML or as JSON—none of which are the server’s internal representation.
A response for a resource contains all the information needed to manipulate that resource
A response should include information about how to process the response
Just like navigating a webpage, a response should contain data about how to proceed with manipulating the data
It is rare to find an API that lives up to REST completely, so RESTful is often used to describe APIs that generally stick to REST principles
Most modern APIs use HTTP methods, pass data back and forth via JSON (this is not a REST requirement!) and is generally stateless
We want to refactor the api to have a Users
table and Listings
table in the database.
We also want to have a listings
endpoint and a users
endpoint.
We also want to be able to see a listing’s user by going to /listing/1/user
and vice-versa /user/1/listings
shows the users listings