App with fastapi, Arangodb and Graphql

Configuring fastapi, arangodb and graphql with authentication:

It uses fastapi graphene, graphene-pydantic, python-arango and fastapi-jwt-auth

mkdir fastapi_learn_project

cd fastapi_learn_project

virtualenv -p python3 env

source env/bin/activate

pip3 install fastapi uvicorn graphene graphene-pydantic==0.1.0 python-arango fastapi-jwt-auth

Setup database

database.py

from arango import ArangoClient

# Initialize the ArangoDB client.
client = ArangoClient(hosts='http://localhost:8529')

# Connect to "test" database as root user.
db = client.db('fastapidemo', username='root', password='****')

You can find how to set up arangodb at:

serializers.py

from typing import List, Optional

from graphene_pydantic import PydanticInputObjectType, PydanticObjectType
from pydantic import BaseModel


class UserModel(BaseModel):
    _id: str
    name: str
    surname: str
    alive: bool
    traits: List[str]


class UserGrapheneModel(PydanticObjectType):
    class Meta:
        model = UserModel


class UserGrapheneInputModel(PydanticInputObjectType):
    class Meta:
        model = UserModel

schema.py

import graphene
from database import db
from typing import List

from serializers import (
    UserGrapheneInputModel,
    UserGrapheneModel,
    UserModel,
)



class Query(graphene.ObjectType):
    all_users = graphene.List(UserGrapheneModel)

    @staticmethod
    def resolve_all_users(parent, info):
   
        #info.context['request'].state.authorize.jwt_required()
        
        cursor = db.aql.execute(
         'FOR s IN user RETURN s'   
        )
        usermodels = [UserModel(**document) for document in cursor]
        return usermodels


class CreateUser(graphene.Mutation):
    class Arguments:
        user_details = UserGrapheneInputModel()

    Output = UserGrapheneModel

    @staticmethod
    def mutate(parent, info, user_details):
        cursor = db.aql.execute(
            'INSERT {name:@name, surname: @surname, alive: @alive, traits: @traits} INTO "user" RETURN NEW',
            bind_vars = user_details)
        user = [document for document in cursor]
        print(user)
        usermodel = UserModel(**user[0])
        return usermodel

class Mutation(graphene.ObjectType):
    create_user = CreateUser.Field()

main.py

import graphene
from fastapi import FastAPI, HTTPException, Depends, Request, APIRouter
from fastapi.responses import JSONResponse
from fastapi_jwt_auth import AuthJWT
from fastapi_jwt_auth.exceptions import AuthJWTException
from pydantic import BaseModel
from starlette.graphql import GraphQLApp
from starlette.datastructures import URL

from schema import Query, Mutation

app = FastAPI()
router = APIRouter()

class User(BaseModel):
    username: str
    password: str

class Settings(BaseModel):
    authjwt_secret_key: str = "secret"
    authjwt_token_location: set = {"cookies"}
    authjwt_cookie_secure: bool = False
    authjwt_cookie_csrf_protect: bool = False

@AuthJWT.load_config
def get_config():
    return Settings()

@app.exception_handler(AuthJWTException)
def authjwt_exception_handler(request: Request, exc: AuthJWTException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"detail": exc.message}
    )

@app.post('/login')
def login(user: User, Authorize: AuthJWT = Depends()):
    if user.username != "test" or user.password != "test":
        raise HTTPException(status_code=401,detail="Bad username or password")

    # subject identifier for who this token is for example id or username from database
    access_token = Authorize.create_access_token(subject=user.username)
    refresh_token = Authorize.create_refresh_token(subject=user.username)
    Authorize.set_access_cookies(access_token)
    Authorize.set_refresh_cookies(refresh_token)
    return {"access_token": access_token}

@app.post('/refresh')
def refresh(Authorize: AuthJWT = Depends()):
    Authorize.jwt_refresh_token_required()

    current_user = Authorize.get_jwt_subject()
    new_access_token = Authorize.create_access_token(subject=current_user)
    # Set the JWT and CSRF double submit cookies in the response
    Authorize.set_access_cookies(new_access_token)
    return {"msg":"The token has been refresh"}

@app.delete('/logout')
def logout(Authorize: AuthJWT = Depends()):
    """
    Because the JWT are stored in an httponly cookie now, we cannot
    log the user out by simply deleting the cookie in the frontend.
    We need the backend to send us a response to delete the cookies.
    """
    Authorize.jwt_required()

    Authorize.unset_jwt_cookies()
    return {"msg":"Successfully logout"}

@app.get('/user')
def user(Authorize: AuthJWT = Depends()):
    Authorize.jwt_required()

    current_user = Authorize.get_jwt_subject()
    return {"user": current_user}

graphql_app = GraphQLApp(schema=graphene.Schema(query=Query, mutation=Mutation))


# Access graphiql ide here
@app.get('/graphiql')
async def graphiql(request: Request):
    request._url = URL('/graphql')
    return await graphql_app.handle_graphiql(request=request)


# request api from frontend here
@app.post("/graphql")
async def graph(request: Request, Authorize: AuthJWT = Depends() ):
    request.state.authorize = Authorize
    return await graphql_app.handle_graphql(request=request)

To run the app:

uvicorn main:app --reload

Search This Blog

Code snippets

App with fastapi, Arangodb and Graphql

Comments

Post a Comment

Popular posts from this blog

Auto Cross covariance Python

Parker Method for Prediction of surface sites