Hi 🙂🖐
Welcome to a new post, Today I will share with you how to create a simple user authentication system in FastAPI using Python.
I will make it easy as much as I can.
[1] - You need to install fastAPI, PyJWT
pip install fastapi
pip install PyJWT
[2] - Download Postman
Postman is an amazing software that allows you to work and deal with APIs.
https://www.postman.com/downloads/
[3] - Import modules
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
import jwt
import json
[4] - Create users.json File
{
"users": [
{
"user_name": "davidramirez",
"password": ")Y1F+Pk5%1",
"user_id": "abc57198-65d0-482f-bba2-4858f1e39b26"
},
{
"user_name": "larsondarren",
"password": "(5aZ5ZwnD*",
"user_id": "880005ba-a590-4592-b8b5-0f54bf29f854"
}
]
}
I create these users using a module called Faker.
If you want to learn more about Faker visit this link: https://dev.to/amr2018/how-to-generates-fake-data-in-python-2d4f
[5] - Define the algorithm and secret
we need in the JWT encoding and decoding process.
algorithm = 'HS256'
secret = '45afs4564f654fsa654f65fsafs'
[6] - instantiate FastAPI class and create the outh2 scheme and load users' data
app = FastAPI()
outh_2_scheme = OAuth2PasswordBearer(tokenUrl = '/signin')
users_db = json.load(open('users.json', 'r'))
[7] - Create Find user function we need this function in signin process
def find_user(user_name, password):
for user in users_db['users']:
if user['user_name'] == user_name and user['password'] == password:
return user
Create Signin route
@app.post('/signin')
async def signin(form_data : OAuth2PasswordRequestForm = Depends()):
user = find_user(form_data.username, form_data.password)
if user:
## generate jwt token
payload = {'user_id': user['user_id']}
token = jwt.encode(payload, secret, algorithm)
return {'access_token': token}
raise HTTPException(
status_code = status.HTTP_401_UNAUTHORIZED,
detail = 'Incorrect username or password'
)
[8] - Create a function called get_current_user
this function gets a token from the outh2 scheme
and decode it to extract the user_id and to find the user with it.
async def get_current_user(token : str = Depends(outh_2_scheme)):
## decode the token
try:
user_id = jwt.decode(token, secret, algorithm)['user_id']
for user in users_db['users']:
if user['user_id'] == user_id:
return user
except:
raise HTTPException(status_code = status.HTTP_401_UNAUTHORIZED)
[9] - Create Me route
Me Is Protected route which means that you need to authenticate
@app.get('/me')
async def me(user = Depends(get_current_user)):
return {'user_name': user['user_name']}
This route depends on the get_current_user function, as we know this function will get the token from the outh2 scheme and decode it to extract the user ID that we need to find the user and return it. Depends in the me route will get the returned user from "get_current_user."
Now we are ready to test the API
Open Postman and copy any user name and password to use in the signing process.
As you can see this is what we need. now we have an access_token
copy this access_token and open a new tap
paste the me route
select GET as the method
select OAuth2.0 as the Authentication type
Now paste the token
Check if it working by removing the token 🙃
full code
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm, OAuth2PasswordBearer
import jwt
import json
algorithm = 'HS256'
secret = '45afs4564f654fsa654f65fsafs'
app = FastAPI()
outh_2_scheme = OAuth2PasswordBearer(tokenUrl = '/signin')
users_db = json.load(open('users.json', 'r'))
def find_user(user_name, password):
for user in users_db['users']:
if user['user_name'] == user_name and user['password'] == password:
return user
@app.post('/signin')
async def signin(form_data : OAuth2PasswordRequestForm = Depends()):
user = find_user(form_data.username, form_data.password)
if user:
## generate jwt token
payload = {'user_id': user['user_id']}
token = jwt.encode(payload, secret, algorithm)
return {'access_token': token}
raise HTTPException(
status_code = status.HTTP_401_UNAUTHORIZED,
detail = 'Incorrect username or password'
)
async def get_current_user(token : str = Depends(outh_2_scheme)):
## decode the token
try:
user_id = jwt.decode(token, secret, algorithm)['user_id']
for user in users_db['users']:
if user['user_id'] == user_id:
return user
except:
raise HTTPException(status_code = status.HTTP_401_UNAUTHORIZED)
@app.get('/me')
async def me(user = Depends(get_current_user)):
return {'user_name': user['user_name']}
Now we're done 🤗
Don't forget to like and follow 🙂
Support me on PayPal 🤗
https://www.paypal.com/paypalme/amr396
Top comments (0)