How to use AsyncSCIMClient with check_server?
#29
Comments
|
Hi. I am not sure how (and when) to tackle this, and make the tool both sync and async without too much code/logic duplication. Anyway, any help is welcome!
Great! Is this something you may want to contribute upstream? Would be awesome to implement ASGI in scim2-server |
I'd love to, except currently my implementation suffers from exactly the problem you describe: too much code/logic duplication. It wouldn't be ideal for your library to be split along fastapi + ASGI / werkzeug + WSGI, with all of the logic agnostic to that also duplicated :/ To clarify, I manually rewrote this, and verified that the important logic is unaffected. However, it's essentially a duplicate of SCIMProvider, which uses an async backend # provider.py
import httpx
from fastapi import HTTPException, Request, Response
from fastapi.responses import JSONResponse
from pydantic import ValidationError
from scim2_models import ...
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from werkzeug.http import unquote_etag # pure function doesn't need to be rewritten, doesn't necessarily need to be replaced, but ideally would be replaced with an equivalent fastapi import
def continue_etag(request: Request, resource) -> bool:
"""Given a request and a resource, checks whether the ETag matches and allows continuing with the request.
If the HTTP header "If-Match" is set, the request may only
continue if the ETag matches. If the HTTP header "If-None-Match"
is set, the request may only continue if the ETag does not
match.
Ported from scim2_server/provider.py's werkzeug implementation to fastapi.
"""
cont = True
resource_version, _ = unquote_etag(resource.meta.version)
if_none_match = request.headers.get("if-none-match")
if if_none_match:
tokens = [token.strip() for token in if_none_match.split(",")]
if resource_version in tokens:
cont = False
if_match = request.headers.get("if-match")
if if_match:
tokens = [token.strip() for token in if_match.split(",")]
if resource_version not in tokens:
cont = False
return cont
...
class CloudSCIMProvider:
"""Provides SCIM protocol functionality.
All functionality is ported from scim2_server.provider.SCIMProvider's werkzeug implementation to fastapi,
with minimal modifications.
"""
def __init__(self, backend: CloudSCIMBackend):
self.backend = backend
self.page_size = 50
self.log = logging.getLogger(self.__class__.__name__)
....
async def call_single_resource(
self, request: Request, resource_endpoint: str, resource_id: str, **kwargs
) -> Response:
find_endpoint = "/" + resource_endpoint
resource_type = self.backend.get_resource_type_by_endpoint(find_endpoint)
if not resource_type:
raise HTTPException(status_code=httpx.codes.NOT_FOUND)
match request.method:
case "GET":
if resource := await self.backend.get_resource(
resource_type.id, resource_id
):
if continue_etag(request, resource):
response_args = self.get_attrs_from_request(request)
self.adjust_location(request, resource)
return make_response(
resource.model_dump(
scim_ctx=Context.RESOURCE_QUERY_RESPONSE,
**response_args,
)
)
else:
return make_response(None, status=httpx.codes.NOT_MODIFIED)
raise HTTPException(status_code=httpx.codes.NOT_FOUND)# api.py
backend = CloudSCIMBackend()
scim_provider = CloudSCIMProvider(backend)
for schema in load_default_schemas().values():
scim_provider.register_schema(schema)
for resource_type in load_default_resource_types().values():
scim_provider.register_resource_type(resource_type)
app = FastAPI(middleware=[Middleware(SCIMMiddleWare, backend=backend)])
...
@app.get("/{resource_endpoint}")
async def resource_get(request: Request, resource_endpoint: str):
return await scim_provider.call_resource(
request, resource_endpoint=resource_endpoint
)# middleware.py
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
# `set_auth_context` is an async context manager that uses ContextVar to make auth_context
# available to every part of the code that needs it e.g. provider, backend, without needing to
# explicitly pass it down in the constructor/function args
class SCIMMiddleWare(BaseHTTPMiddleware):
def __init__(self, app, backend: CloudSCIMBackend):
super().__init__(app)
self.backend = backend
self.log = logging.getLogger(self.__class__.__name__)
async def check_auth(self, request: Request):
# this function implementation is highly specific to our backend
pass
async def dispatch(self, request: Request, call_next: RequestResponseEndpoint):
try:
# TODO redirect URLs that end with .scim to the non-suffixed URL
auth_context = None
if "ServiceProviderConfig" not in request.url.path:
auth_context = await self.check_auth(request)
async with self.backend, set_auth_context(auth_context):
response = await call_next(request)
if "location" not in response.headers:
response.headers["location"] = str(request.url)
return response
except HTTPException as e:
self.log.exception(e)
return make_error(Error(status=e.status_code, detail=e.detail))
except SCIMException as e:
self.log.exception(e)
return make_error(e.scim_error)
except ValidationError as e:
self.log.exception(e)
return make_error(Error(status=httpx.codes.BAD_REQUEST, detail=str(e)))
except Exception as e:
self.log.exception(e)
sentry_sdk.capture_exception(e)
return make_error(Error(status=httpx.codes.INTERNAL_SERVER_ERROR))I am not aware of solid and well-tested strategies for having one implementation provide both an async and sync interface. In my experience, However, if you are interested in the complete UPDATE: I am also finding some problems with my implementation as-is, for example exception handling works differently in |
Uh oh!
There was an error while loading. Please reload this page.
Hello, thanks for writing a great series of libraries! I am having trouble understanding how to use the
check_serverfunctionality, while passing in theAsyncSCIMClient. My hunch is that this is not possible, because thecheck_serverfunction flow seems to only be implemented synchronously. Is there anyway to get the benefit of the check_server test for my scim implementation, using an async client (and using the in-memory app object rather than hitting a server over network)?Some more background: we have an ASGI FastAPI app. I originally tried mounting the
SCIMProviderfromscim2_serverasWSGIMiddlewareonto my FastAPI app, but since our app + DB layer is all async, I had to useasync_to_syncin a bunch of places, and I ended up getting errors/issues with the event loop due to that. So instead, I opted to rewrite the SCIMProvider using an async interface (I replacedwerkzeugwithfastapiandstarlette), while keeping all of the SCIM protocol logic + interfaces the same, such that I can use the same Backend interface, and leverage all the awesome work you folks have done with implementing SCIM. However, now I am having trouble finding good ways to leverage your library to also test the implementation, hence this issue.Any help/guidance is much appreciated, thanks!
The text was updated successfully, but these errors were encountered: