Interactive Lecture Week 3

RESTful is an architectural style for building out the urls of an application server. It specifies which HTTP methods should be used according to a desired CRUD operation.

CRUD Operations

CRUD Mappings

The remainder of this session will continue from the library rental application from last week. In this session, we'd be moving away from the CLI and back into implementing web functionality. The starter repository can be found here

Just like we learnt back in Week 1 GET routes are specified with @app.get('<route>'). In week 1, we declared the GET route directly on the FastAPI app itself however, we'd be following the convention from now on and declaring our routes on an APIRouter. As such:

GET requests can be implemented with @<router>.post('<route>')
POST requests can be implemented with @<router>.post('<route>')
PUT requests can be implemented with @<router>.put('<route>')
DELETE requests can be implemented with @<router>.delete('<route>')

For example if we have the following code

from fastapi import APIRouter

user_router = APIRouter()

Then, the following would be ways of declaring route functions on that router

@user_router.get("/users")
def get_users():
    # Some implementation here that responds to a GET request on the endpoint /user
    pass
    
@user_router.get("/user/{id}")
def get_user_by_id(id:int):
    # Some implementation here that responds to a GET request on the endpoint /user
    pass

@user_router.post("/users")
def create_user():
    # Some implementation here that responds to a POST request on the endpoint /user
    pass

@user_router.put("/users")
def update_profile():
    # Some implementation here that responds to a PUT request on the endpoint /user
    pass

@user_router.delete("/users")
def delete_user():
    # Some implementation here that responds to a DELETE request on the endpoint /user
    pass

In larger applications with complex functionality routes are split based on the model or the functionality. i.e. If we were to have the Todo app we built in the lab this week. We would

Have the routes for authenticating in one file app/routers/auth.py
Have the routes for the CRUD functionality for managing Todos in another file app/routers/todo.py
Have the routes for user management in another file app/routers/user.py

Each router declared in the files above would then be linked to a main router that's used in the app by using code similar to the below.

from .users import user_router
main_router.include_router(user_router)

Most routes interact with the database in production apps. We're going to use FastAPI's dependency injection functionality to get a reference to the database session for each route. The implementation of this isn't necessary to know, just know that if you need to perform some operation on data in the database, you can include a reference to a Depends object in the arguments of the function and FastAPI will inject that into the function and make it available for you. The implementation of SessionDep can be found in app/database.py

from app.database import SessionDep

@user_router.post("/user")
async def a_function(db: SessionDep ):
    # Functionality redacted here for brevity
    pass

Implementation

Let's implement the functions above.

As our app gets larger, we'd want to specify and standardize the parameters our routes accept, and the response types our application returns. This is done by using Data Schemas

Data Schemas are defined using Pydantic models or SQLModel. In this course we'd use SQLModel to specify our data schemas. (SQLModel extends Pydantic Models, so essentially they're the same). Data schemas specify the structure, data types, and validation rules for the data your API handles. If you have experience with flask, a similar library that provided this functionality was marshmallow

Data Schemas are used to:

Validate request data
Serialize response data
Generate standardized OpenAPI documentation.

When attempting this at your own time, copy and paste the code below into a file app/models.py. Note that this class does not include table=True so it will NOT create a table in our database, it would be used purely for validation

# Example code for a UserCreate class that specifies fields 
# a user should submit when registering

from pydantic import EmailStr  #insert at top of the file

class UserCreate(SQLModel):
    username:str
    email: EmailStr = Field(max_length=255)
    password: str = Field(min_length=8, max_length=128)

In many applications it is typical to restrict access to features from unauthorized parties. If your application deals with user data then users should be able to only manipulate the data that belong to them. This is achieved by applying the following concepts

Authentication: Is validating that the user is who they claim to be. It refers to a user's ability to prove who they say they are. This is usually done by proving credentials (username and password)

Authorization: Controlling what a particular user class (role) can access. This is usually managed by RBAC (role-based access control) and it limits the operations a user can perform.

Example

A person in the university can have one of many roles

A Student: Has the ability to log in (authenticate) and access myElearning.

A Staff: Has the ability to log in (authenticate) and access myElearning.

A Student: has the abililty to view content in a course they are registered in, however they do NOT have the ability to modify the content in the course shell or view the data on other students (unauthorized)

A Staff: has the ability to view and edit content in the courses they have the rights to modify as well as see data on their registered students (authorized)

We'd implement OAuth2 with Password (and hashing) using JWT Bearer tokens.

Hashing means converting some content (a password in this case) into a string that looks like gibberish that can't be easily reversed. Verification of passwords are done by passing exactly the same password to get exactly the same gibberish.

But you cannot convert from the gibberish back to the password.

Create a new model in app/models.py as follows

class Token(SQLModel):
    access_token: str
    token_type: str

Create a new file app/routers/auth.py lets create a new route for authentication

from fastapi import APIRouter, HTTPException, Depends
from sqlmodel import select
from app.database import SessionDep
from app.models import *
from app.auth import verify_password, create_access_token
from fastapi.security import OAuth2PasswordRequestForm
from typing import Annotated
from fastapi import status

auth_router = APIRouter()

@auth_router.post("/token")
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: SessionDep
) -> Token:
    user = db.exec(select(User).where(User.username == form_data.username)).one_or_none()
    if not user or not verify_password(plaintext_password=form_data.password, encrypted_password=user.password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
        
    access_token = create_access_token(data={"sub": user.username},)

    return Token(access_token=access_token, token_type="bearer")

Securing your endpoints to ensure that a user MUST be authenticated to call it is as simple as using the provided Dependency AuthDep from auth.py by including it in the list of arguments of the route function.

@user_router.get("/users", response_model=list[UserResponse])
def get_users(db: SessionDep, user:AuthDep):
    all_users = db.exec(select(User)).all()
    return all_users

The endpoints you develop can be tested using the built in tester by visiting the URL of your app with the path /docs e.g. http://localhost:8080/docs

The application can also be tested with Postman

app/models.py

from sqlmodel import Field, SQLModel
from typing import Optional
from pydantic import EmailStr  #insert at top of the file

class UserUpdate(SQLModel):
    username: Optional[str] = None
    email: Optional[EmailStr] =None

class UserCreate(SQLModel):
    username:str
    email: EmailStr = Field(max_length=255)
    password: str = Field(min_length=8, max_length=128)

class UserResponse(SQLModel):
    id: Optional[int]
    username:str
    email: EmailStr

class Token(SQLModel):
    access_token: str
    token_type: str


class User(SQLModel, table=True):
    id: Optional[int] =  Field(default=None, primary_key=True)
    username:str = Field(index=True, unique=True)
    email:str = Field(index=True, unique=True)
    password:str

app/routers/__init__.py

from fastapi import APIRouter
main_router = APIRouter()

from .user import user_router
main_router.include_router(user_router)

from .auth import auth_router
main_router.include_router(auth_router)

app/routers/auth.py

from fastapi import APIRouter, HTTPException, Depends
from sqlmodel import select
from app.database import SessionDep
from app.models import *
from app.auth import verify_password, create_access_token
from fastapi.security import OAuth2PasswordRequestForm
from typing import Annotated
from fastapi import status

auth_router = APIRouter(tags=["Authentication"])

@auth_router.post("/token")
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: SessionDep
) -> Token:
    user = db.exec(select(User).where(User.username == form_data.username)).one_or_none()
    if not user or not verify_password(plaintext_password=form_data.password, encrypted_password=user.password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
        
    access_token = create_access_token(data={"sub": user.username},)

    return Token(access_token=access_token, token_type="bearer")

app/router/user.py

from fastapi import APIRouter, HTTPException
from app.database import SessionDep
from sqlmodel import select 
from app.models import User, UserCreate, UserResponse, UserUpdate
from app.auth import encrypt_password, AuthDep

user_router = APIRouter(tags=["User Management"])

@user_router.get("/users", response_model=list[UserResponse])
def get_users(db: SessionDep, user:AuthDep):
    all_users = db.exec(select(User)).all()
    return all_users
    
@user_router.get("/user/{id}", response_model=UserResponse)
def get_user_by_id(id:int, db: SessionDep, user:AuthDep):
    user = db.get(User, id)
    if not user:
        raise HTTPException(status_code=404)
    return user

@user_router.post("/users")
def create_user(db: SessionDep, user_data:UserCreate):
    user = User(username=user_data.username, email=user_data.email,)
    user.password = encrypt_password(user_data.password)
    db.add(user)
    db.commit()
    return user
    user = db.get(User, id)
    if not user:
        raise HTTPException(status_code=404)

@user_router.put("/user/{id}", response_model=UserResponse)
def update_profile(id:int, user_data:UserUpdate, db: SessionDep, user:AuthDep):

    if user_data.username:
        user.username = user_data.username
    if user_data.email:
        user.email = user_data.email
    
    db.add(user)
    db.commit()
    return user


@user_router.delete("/user/{id}")
def delete_user(id:int, db: SessionDep, user:AuthDep):
    user = db.get(User, id)
    if not user:
        raise HTTPException(status_code=404)
    
    db.delete(user)
    db.commit()
    return True