Skip to content

Instantly share code, notes, and snippets.

@aherok
Created March 4, 2020 11:26
Show Gist options
  • Save aherok/9f3e555519064bc100fd2caa866fd451 to your computer and use it in GitHub Desktop.
Save aherok/9f3e555519064bc100fd2caa866fd451 to your computer and use it in GitHub Desktop.
Starlette + Ariadne handling sessions test
"""
Required to install:
* starlette
* ariadne
* uvicorn
run via `uvicorn app:app`
"""
import logging
from dataclasses import dataclass
from ariadne import QueryType, gql, make_executable_schema, MutationType
from ariadne.asgi import GraphQL
from starlette.applications import Starlette
from starlette.authentication import AuthenticationBackend, AuthCredentials, UnauthenticatedUser
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.middleware.sessions import SessionMiddleware
from starlette.requests import HTTPConnection, Request
from starlette.responses import PlainTextResponse
from starlette.routing import Route
log = logging.getLogger(__name__)
#
# AUTH
#
ALLOWED_USERNAMES = ['admin']
@dataclass
class User:
username: str
def authenticate(username: str):
if username in ALLOWED_USERNAMES:
return User(username=username)
return False
def login(request: Request, user: User):
log.info(f"setting session: {user.username}")
request.cookies['username'] = user.username
request.session['username'] = user.username
return True
#
# AUTH BACKEND
#
class AuthBackend(AuthenticationBackend):
async def authenticate(self, request: HTTPConnection):
log.info(f"COOKIE: {request.cookies}")
# AUTH LOGIC WILL GO HERE
#
# user = await get_user_from_session(request)
# if not user:
# return None
#
# return AuthCredentials(["authenticated"]), user
#
# for now let's just return defaul anonymous user
return AuthCredentials(["anonymous"]), UnauthenticatedUser()
#
# GRAPHQL
#
type_defs = gql("""
type Query {
getHeaders: String!
}
type Mutation {
login(username: String!): Boolean!
logout: Boolean!
}
""")
query = QueryType()
@query.field("getHeaders")
def resolve_hello(_, info):
request = info.context["request"]
cookies = ", ".join(request.cookies.keys())
return "Cookies: %s --END" % cookies
mutation = MutationType()
@mutation.field("login")
def resolve_login(_, info, username):
request = info.context["request"]
user = authenticate(username)
if user:
login(request, user)
return True
return False
@mutation.field("logout")
def resolve_logout(_, info):
request = info.context["request"]
request.session.pop('username')
return False
schema = make_executable_schema(type_defs, [query, mutation])
#
# ROUTES
#
async def homepage(request: HTTPConnection):
output = f"""Hello world!
SESSION: {request.session}
COOKIE: {request.cookies}
"""
response = PlainTextResponse(output)
return response
#
# APP
#
middleware = [
Middleware(SessionMiddleware, secret_key="SECRETKEY", session_cookie='sessid'),
Middleware(AuthenticationMiddleware, backend=AuthBackend()),
]
app = Starlette(
debug=True, routes=[Route("/", homepage)], middleware=middleware
)
app.mount("/graphql", GraphQL(schema, debug=True))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment