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

1
2
3
4
5
6
7
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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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

Comments

Post a Comment

Popular posts from this blog

Auto Cross covariance Python

-
Auto Cross Covariance for converting protein sequence descriptor to single length vector import numpy as np # z1 z2 and z3 descriptor was used to represent the protein sequence # Index j was used for the z-scales (j = 1, 2, 3), # n is the number of amino acids in a sequence, # index i is the amino acid position (i = 1, 2, ...n) # l is the lag (l = 1, 2, ...L). # a short range of lags (L= 1, 2, 3, 4, 5) Z = np.random.rand(3,80) print(Z) #Z = np.array([[1,2,3,4],[5,6,7,8],[9,10,11,12]]) n = Z.shape[1] n = n-1 print(n) # Autocovariance column = [] for j in range(0,3): row = [] for l in range(0,5): summ = 0 for i in range(0,n-l): rightsum = (Z[j,i]*Z[j,i+1])/(n-l) summ = summ + rightsum row.append(summ) column.append(row) R = np.array(column) print(R) # Cross Covariance ja = [0,1,2,0,1,2] ka = [1,0,0,2,2,1] column = [] for j,k in zip(ja,ka): row = [] for l in range(0,5): summ = 0 for i in range...
Read more

Generating UUID4 from custom random number generator (RNG) using rust

-
First you need to generate random seed: It can be from any RNG generator, for example in near protocol sdk,  https://docs.rs/near-sdk/0.11.0/near_sdk/env/fn.random_seed.html pub fn random_seed() -> Vec<u8> use rand::Rng; fn main () { let mut rng = rand::thread_rng(); let mut randvec: Vec < u8 > = Vec ::new(); let mut counter = 0 ; let result = loop { counter += 1 ; let n1: u8 = rng.gen(); randvec.push(n1); if counter == 32 { break randvec; } }; println! ( "{:?}" , result); } Its generate vector, but you need an array to feed into from_seed function. This is the complete code. use rand::{rngs::StdRng, RngCore, SeedableRng, Rng}; use uuid::{Builder, Variant, Version}; fn main () { let mut rng = rand::thread_rng(); let mut randvec: Vec < u8 > = Vec ::new(); let mut counter = 0 ; let result_vec = loop { counter ...
Read more