Module neuroio.auth_token
None
None
View Source
import typing
from httpx import Auth, Request, Response
class AuthorizationTokenAuth(Auth):
"""Describes an API Token requests authentication."""
def __init__(self, api_token: str, header_name: str = None):
"""
:param api_token: The API token that will be sent.
:param header_name: Name of the header field.
"""
self.api_token = api_token
self.header_name = header_name or "Authorization"
def auth_flow(
self, request: Request
) -> typing.Generator[Request, Response, None]:
request.headers[self.header_name] = f"Token {self.api_token}"
yield request
Classes
AuthorizationTokenAuth
class AuthorizationTokenAuth(
api_token: str,
header_name: str = None
)
View Source
class AuthorizationTokenAuth(Auth):
"""Describes an API Token requests authentication."""
def __init__(self, api_token: str, header_name: str = None):
"""
:param api_token: The API token that will be sent.
:param header_name: Name of the header field.
"""
self.api_token = api_token
self.header_name = header_name or "Authorization"
def auth_flow(
self, request: Request
) -> typing.Generator[Request, Response, None]:
request.headers[self.header_name] = f"Token {self.api_token}"
yield request
Ancestors (in MRO)
- httpx.Auth
Class variables
requires_request_body
requires_response_body
Methods
async_auth_flow
def async_auth_flow(
self,
request: httpx.Request
) -> AsyncGenerator[httpx.Request, httpx.Response]
Execute the authentication flow asynchronously.
By default, this defers to .auth_flow(). You should override this method
when the authentication scheme does I/O and/or uses concurrency primitives.
View Source
async def async_auth_flow(
self, request: Request
) -> typing.AsyncGenerator[Request, Response]:
"""
Execute the authentication flow asynchronously.
By default, this defers to `.auth_flow()`. You should override this method
when the authentication scheme does I/O and/or uses concurrency primitives.
"""
if self.requires_request_body:
await request.aread()
flow = self.auth_flow(request)
request = next(flow)
while True:
response = yield request
if self.requires_response_body:
await response.aread()
try:
request = flow.send(response)
except StopIteration:
break
auth_flow
def auth_flow(
self,
request: httpx.Request
) -> Generator[httpx.Request, httpx.Response, NoneType]
Execute the authentication flow.
To dispatch a request, yield it:
yield request
The client will .send() the response back into the flow generator. You can
access it like so:
response = yield request
A return (or reaching the end of the generator) will result in the
client returning the last response obtained from the server.
You can dispatch as many requests as is necessary.
View Source
def auth_flow(
self, request: Request
) -> typing.Generator[Request, Response, None]:
request.headers[self.header_name] = f"Token {self.api_token}"
yield request
sync_auth_flow
def sync_auth_flow(
self,
request: httpx.Request
) -> Generator[httpx.Request, httpx.Response, NoneType]
Execute the authentication flow synchronously.
By default, this defers to .auth_flow(). You should override this method
when the authentication scheme does I/O and/or uses concurrency primitives.
View Source
def sync_auth_flow(
self, request: Request
) -> typing.Generator[Request, Response, None]:
"""
Execute the authentication flow synchronously.
By default, this defers to `.auth_flow()`. You should override this method
when the authentication scheme does I/O and/or uses concurrency primitives.
"""
if self.requires_request_body:
request.read()
flow = self.auth_flow(request)
request = next(flow)
while True:
response = yield request
if self.requires_response_body:
response.read()
try:
request = flow.send(response)
except StopIteration:
break