Continuing with the FastAPI OAuth2 scopes topic, in this installment of our Python FastAPI learning series, we will implement seven (7) new routes that perform CRUD operations on the employees table. These new routes require scopes that we have implemented but have not used so far: user:write, admin:read, and admin:write. Please recall that we proposed this implementation in the last post.

🐍 Index of the Complete Series.

126-feature-image.png
Python FastAPI: OAuth2 Scopes Part 03 - New CRUD Endpoints and User-Assigned Scopes

The code requires Python 3.12.4. Please refer to the following discussion on how to upgrade to Python 3.12.4.

🚀 Please note, complete code for this post can be downloaded from GitHub with: 

git clone -b v0.12.0 https://github.com/behai-nguyen/fastapi_learning.git

❶ We have added a few more files to the application. The updated structure of the project is outlined below.

-- Please note, those marked with ★ are updated, and those marked with ☆ are new. 

/home/behai/fastapi_learning/
.
├── cert
│   ├── cert.pem
│   └── key.pem
├── .env
├── logger_config.yaml
├── main.py ★
├── pyproject.toml ★
├── pytest.ini ★
├── README.md ★
├── sql_scripts
│   └── postgres
│       └── 04_test_get_employees_stored_method.sql ☆
├── src
│   └── fastapi_learning
│       ├── businesses
│       │   ├── app_business.py
│       │   ├── base_business.py
│       │   ├── base_validation.py ★
│       │   ├── employees_mgr.py ★
│       │   └── employees_validation.py ★
│       ├── common
│       │   ├── consts.py ★
│       │   ├── jwt_utils.py ★
│       │   ├── queue_logging.py
│       │   └── scope_utils.py
│       ├── controllers
│       │   ├── admin.py
│       │   ├── auth.py ★
│       │   ├── employees_admin.py ☆
│       │   ├── __init__.py ★
│       │   └── required_login.py ☆
│       ├── __init__.py ★
│       ├── models
│       │   └── employees.py ★
│       ├── static
│       │   └── js
│       │       └── application.js ☆
│       └── templates
│           ├── admin
│           │   └── me.html
│           ├── auth
│           │   ├── home.html ★
│           │   └── login.html
│           ├── base.html ★
│           └── emp ☆
│               ├── insert.html
│               ├── search.html
│               ├── search_result.html
│               └── update.html
└── tests
    ├── business
    │   ├── test_employees_mgr.py ★
    │   ├── test_employees_validation.py ☆
    │   └── test_scope_utils.py
    ├── conftest.py
    ├── __init__.py ★
    ├── integration
    │   ├── test_admin_itgt.py
    │   ├── test_api_itgt.py
    │   ├── test_auth_itgt.py
    │   ├── test_employees_itgt.py ☆
    │   ├── test_expired_jwt.py
    │   ├── test_scope_permission_itgt.py
    │   └── test_scope_ui_itgt.py
    ├── README.md
    └── unit
        └── test_employees.py ★

❷ Update a PostgreSQL stored method.

Please recall from the seventh post that the database used is a modified version of the one used in the two examples provided for the bh-database wrapper classes for SQLAlchemy. As such, the stored methods created by these SQL scripts should be present in the database. For this learning application, we have added the email and password columns to the employees table, making the PostgreSQL stored method get_employees(...) outdated.

We need to apply this new PostgreSQL script sql_scripts/postgres/04_test_get_employees_stored_method.sql to update the mentioned stored method. You might need to update the schema name to your appropriate one.

🙏 I maintain two separate databases: one for the bh-database's examples and another for this application.

❸ Let's discuss the CRUD rules for our implementation. Please note that user number and employee number are used interchangeably and both refer to the employees.emp_no column.

⓵ employees.email is unique. However, there is no unique database constraint. This rule is enforced at the business level when adding a new employee.

⓶ Updating existing employee records does not include the email and password columns. If they are present in the submitted data, they are simply ignored. This means we cannot update employees.email and employees.password via the front end.

⓷ Logged-in users with user:write scope can update their own record. The logged-in user number must match the submitted user number.

⓸ Logged-in users with admin:write scope can update existing employee records and also create new employee records.

⓹ Please recall from the seventh post that the business layer determines if the submitted data is for updating existing records or inserting new records. In hindsight, this point was not made very clear in the mentioned post.

  • The employees table gets updated when the submitted data has a value for the emp_no column; email and password are ignored even if they are in the submitted data.
  • Inserting a new record takes place when the submitted data has no value for the emp_no column; email and password are required.

❹ The table below lists information on the new routes that we are going to implement:

MethodScopesRouteResponse
GETadmin:read/emp/searchHTML
GET, POSTadmin:read/emp/search/{partial-last-name}/{partial-first-name}HTML, JSON
GETadmin:read/emp/admin-get-update/{emp_no}HTML, JSON
GETuser:read/emp/own-get-update/{emp_no}HTML, JSON
POSTadmin:write/emp/admin-saveJSON
POSTuser:write/emp/user-saveJSON
GETadmin:write/emp/newHTML

Let's describe each of these endpoints.

  1. /emp/search: Get the employee search form.
  2. /emp/search/{partial-last-name}/{partial-first-name}: Perform the employee search based on partial last name and partial first name.
  3. /emp/admin-get-update/{emp_no}: A logged-in user who potentially has the admin:write scope, attempts to retrieve an employee record for editing. The path parameter /{emp_no} identifies the employee. The response is either an HTML page or a JSON object. 💥 This route only requires the admin:read scope because retrieving the data means reading, not yet writing or updating.

  4. /emp/own-get-update/{emp_no}: A logged-in user who potentially has the user:write scope, attempts to retrieve their own record for editing. The path parameter /{emp_no} identifies the employee. The response is either an HTML page or a JSON object. 💥 This route only requires the user:read scope because retrieving the data means reading, not yet writing or updating.

    This route checks that the value of the path parameter /{emp_no} matches the logged-in user number; otherwise, the request is rejected with an error response, even if the logged-in user has the admin:read scope.

  5. /emp/admin-save: A logged-in user attempts to update or insert employee records. This endpoint requires the admin:write scope.

  6. /emp/user-save: A logged-in user attempts to update their own details. This route checks that the value of the submitted parameter empNo matches the logged-in user number; otherwise, the request is rejected with an error response, even if the logged-in user has the user:write scope.

  7. /emp/new: Get the add new employee form.

❺ In this section, we discuss the code changes, which are a bit involved.

⓵ The pyproject.toml file now includes two new packages: python-multipart and email-validator. I have recently installed Ubuntu 24.04.1 LTS (Noble Numbat) and Python 3.13.0, and the application installation requires these two packages. I must have had these two available in the previous environment without realising it.

⓶ Using custom exception handler to manage redirecting to login page. Principal references for this refactoring:

  1. How to redirect to login in FastAPI
  2. FastAPI Install custom exception handlers.

💥 This implementation renders the method def is_logged_in(...) obsolete. However, we will not refactor this part of the code just yet. We will do so in a later post.

● The new controllers/required_login.py module implements the method get_logged_in_user(...), which is the Depends() callable for endpoint handlers. This method raises the custom exception RequiresLogin if the request is not authenticated. If the request authentication is successful, it performs a database query to read the current logged-in user information. Since querying the database is expensive, we could refactor it so that all user information comes from the token payload.

The method get_logged_in_user(...) is a variation of the existing method get_current_user(...). However, they have different responsibilities.

● In the main.py module, we install the exception handler to catch the RequiresLogin exception and redirect as follows: 

109
110
111
112
113

# Redirect to login using custom exception handlers.
# See https://stackoverflow.com/a/76887329
@app.exception_handler(RequiresLogin)
async def requires_login(request: Request, _: Exception):
    return RedirectResponse(url='/auth/login?state=2')

In endpoint handler methods, we call the callable get_logged_in_user(...) as follows: 

@router.get("/own-get-update/{emp_no}")
async def user_update(request: Request, emp_no: str,
                      user = Depends(get_logged_in_user),
                      response = Security(get_employee_to_update, scopes=["user:read"])):

We will retrofit this into endpoint handler methods in the controllers/admin.py and controllers/auth.py modules in the next revision.

⓷ Added a new generic async def verify_user_scopes(...) method to manage user scopes verification. In the next revision, we might refactor endpoint handler methods in the controllers/auth.py module and the controllers/admin.py module to use this new method as well.

⓸ The code around the employees table requires refactoring to support new functionalities. We discuss these changes in this section.

⑴ In the models/employees.py module, we added some new database query functions. They are mostly one-liners. The associated tests/unit/test_employees.py module has also been refactored to include tests for the new methods.

⑵ As apparent from this earlier section, the new functionalities require the user number of the logged-in user. The following changes were required:

● In the TokenData(BaseModel) class, we added a new field user_number: int | None = None.

The JWT-related code was updated to include this user_number: int value.

● In the async def login(...) method, the user number is passed to the token payload after logging in successfully: 

194
195
196

    access_token = create_access_token(data={'sub': op_status.data[0]['email'],
                                             'emp_no': op_status.data[0]['emp_no'],
                                             'scopes': op_status.data.scopes})

● The home page also requires the logged-in user number. The helper method def __home_page(...) was refactored to include the user number in the page context data: 

97

        data.update({"user_number": token_data.user_number, "user_scopes": token_data.scopes})

Finally, when decoding a JWT, the value of the user number is copied into the result TokenData(BaseModel)'s new field user_number: int. The method def decode_access_token(...) was refactored with the following addition: 

44
45
46
47
48

        usernumber: int = int(payload.get("emp_no")) if 'emp_no' in payload else None
        if usernumber is None:
            return credentials_exception
        
        return TokenData(user_name=username, user_number=usernumber, scopes=payload.get("scopes", []))

⑶ The businesses/employees_validation.py module includes two new validation forms: UpdateEmployeeForm and AddEmployeeForm. The implementations of these two new forms are based on the CRUD rules discussed earlier.

The new associated test module tests/business/test_employees_validation.py should have sufficient tests to cover all possible data validation scenarios for the forms.

⑷ As anticipated, there were a number of changes in the businesses/employees_mgr.py module:

● To sufficiently cover all permission scenarios, we added two more entries to the MOCK_USER_SCOPES list:

  1. User weijing.showalter.67000@gmail.com, scope: admin:read. This user can search and read other users' information.
  2. User nidapan.samarati.262556@gmail.com, scope: user:read. This user can only read their own information; they can't update their own details.

Please refer to this section of an earlier post for the first discussion on the mock user list.

● Added two more database query methods: def select_by_partial_last_name_and_first_name(...) and def select_by_employee_number(...).

● Implemented the required override methods to capture, validate, update, and insert data. These methods are quite involved, but they should be readable, as they are quite short and each has a single distinct responsibility.

👉 The associated test module tests/business/test_employees_mgr.py also had more test methods added to cover tests for the new implementations. The test methods should also help in understanding the code properly.

⑸ Finally, the new controllers/employees_admin.py module, which implements all new routes as described previously. Let's take a look at the endpoint handler code for the route /emp/admin-save: 

232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259

async def do_save(request: Request, 
                  security_scopes: SecurityScopes,
                  token: Annotated[str, Depends(oauth2_scheme)]):
    
    res, _, exception = await verify_user_scopes(security_scopes, token, INVALID_PERMISSIONS_MSG)

    if not res:
        return make_500_status(exception.detail).as_dict()

    form = await request.form()    
    return EmployeesManager().write_to_database(form._dict).as_dict()
    
@router.post("/admin-save", response_class=JSONResponse)
async def admin_save(request: Request,
                     user = Depends(get_logged_in_user),
                     response = Security(do_save, scopes=["admin:write"])):
    
    """ 
    Route: https://0.0.0.0:5000/emp/admin-save

    This method is for users with 'admin:write' to write new employee records,
    or to update existing records.

    empNo can be blank, None or completely absent. Fields 'email' and 
    'password' must have values.
    """

    return response

The structure of the code should be familiar to us. I would like to note the following points:

● user = Depends(get_logged_in_user) is the custom exception handler to manage login redirection, which we have discussed earlier.

● We use response = Security(do_save, scopes=["admin:write"]) instead of the already familiar response = Annotated[JSONResponse, Security(do_save, scopes=["admin:write"])]. For an explanation, please refer to this GitHub issue: Cannot use Response type hint as dependency annotation #10127.

● Note that when an exception or an error occurs, for JSON response, we now return a ResultStatus class via return make_500_status(exception.detail).as_dict().

This has always been the intention since the start of this learning application. However, I have failed to realise this implementation in the controllers/admin.py and controllers/auth.py modules. We will refactor this in the next revision.

👉 The associated test module tests/integration/test_employees_itgt.py has more than twenty (20) coverage tests for the new routes. These test methods should aid in understanding the code, too.

⓹ Finally, refactoring of the front-end code or UI:

⑴ Added a new application JavaScript file: static/js/application.js.

⑵ The base HTML template templates/base.html now references more JavaScript files from this GitHub repository. In a production deployment, we don't reference this many JavaScript or CSS files as it would slow down the loading of the pages. Instead, we would minify all files into a single large JavaScript file and a single large CSS file, eliminating network round trips. I have discussed minification in the post JavaScript and CSS minification

⑶ The home page template templates/auth/home.html has also received significant refactoring. As a result of new UI elements being added, the UI elements' state calculations were changed quite significantly compared to the previous revision. The current code is: 

48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

{% set user_read_scope = ['user:read'] %}
{% set user_write_scope = ['user:write'] %}
{% set admin_read_scope = ['admin:read'] %}
{% set admin_write_scope = ['admin:write'] %}

{% set user_read_disabled_str = '' %}
{% set user_write_disabled_str = '' %}
{% set admin_read_disabled_str = '' %}
{% set admin_write_disabled_str = '' %}

{% if not enable_no_scopes_ui() %}
{%     if not has_required_scopes(user_read_scope, data['user_scopes']) %}
{%         set user_read_disabled_str = 'disabled' %}
{%     endif %}

{%     if not has_required_scopes(user_write_scope, data['user_scopes']) %}
{%         set user_write_disabled_str = 'disabled' %}
{%     endif %}

{%     if not has_required_scopes(admin_read_scope, data['user_scopes']) %}
{%         set admin_read_disabled_str = 'disabled' %}
{%     endif %}

{%     if not has_required_scopes(admin_write_scope, data['user_scopes']) %}
{%         set admin_write_disabled_str = 'disabled' %}
{%     endif %}
{% endif %}

Basically, the page needs to know if the logged-in user has the user:read, user:write, admin:read, and admin:write scopes. The previous revision needed only user:read.

⑷ Finally, the four new employee pages templates/emp/*.html. They should be self-explanatory, so we will not go into detailed descriptions.

❻ The following screenshots show some new UI added in this installment:


❼ We conclude this post here. I hope we are not overwhelmed by the length of this post. Before finishing off, I would like to recap the following to-do items:

  1. Retrofit the custom exception handler to manage login redirection, i.e., use the method async def get_logged_in_user(...) in endpoint handler methods in the controllers/admin.py and controllers/auth.py modules. This point has also been discussed in detail earlier.
  2. As discussed previously, we might also retrofit the async def verify_user_scopes(...) method to endpoint handler methods in the controllers/admin.py and controllers/auth.py modules.
  3. As discussed, when an error or exception occurs, and the requested response is JSON, existing endpoint handlers in the controllers/admin.py and controllers/auth.py modules should return a ResultStatus class via calling the make_500_status(...) method.

Thank you for reading. I hope you find the information in this post useful. Stay safe, as always.

✿✿✿

Feature image source:

🐍 Index of the Complete Series.