feat: FastAPI support

pylti1p3/contrib/django/lti1p3_tool_config/models.py

@@ -170,9 +170,9 @@ def to_dict(self):
             "auth_audience": self.auth_audience,
             "key_set_url": self.key_set_url,
             "key_set": json.loads(self.key_set) if self.key_set else None,
-            "deployment_ids": json.loads(self.deployment_ids)
-            if self.deployment_ids
-            else [],
+            "deployment_ids": (
+                json.loads(self.deployment_ids) if self.deployment_ids else []
+            ),
         }
         return data
 

pylti1p3/contrib/fastapi/__init__.py

@@ -0,0 +1,7 @@
+# flake8: noqa
+from .cookie import FastAPICookieService
+from .launch_data_storage.cache import FastAPICacheDataStorage
+from .message_launch import FastAPIMessageLaunch
+from .oidc_login import FastAPIOIDCLogin
+from .request import FastAPIRequest
+from .session import FastAPISessionService

pylti1p3/contrib/fastapi/cookie.py

@@ -0,0 +1,38 @@
+from pylti1p3.cookie import CookieService
+
+
+class FastAPICookieService(CookieService):
+    _request = None
+    _cookie_data_to_set = None
+
+    def __init__(self, request):
+        self._request = request
+        self._cookie_data_to_set = {}
+
+    def _get_key(self, key):
+        return self._cookie_prefix + "-" + key
+
+    def get_cookie(self, name):
+        return self._request.get_cookie(self._get_key(name))
+
+    def set_cookie(self, name, value, exp=3600):
+        self._cookie_data_to_set[self._get_key(name)] = {
+            "value": value,
+            "exp": exp,
+        }
+
+    def update_response(self, response):
+        is_secure = self._request.is_secure()
+        for key, cookie_data in self._cookie_data_to_set.items():
+            cookie_kwargs = {
+                "key": key,
+                "value": cookie_data["value"],
+                "max_age": cookie_data["exp"],
+                "secure": is_secure,
+                "path": "/",
+                "httponly": True,
+                "samesite": None,
+            }
+            if is_secure:
+                cookie_kwargs["samesite"] = "None"
+            response.set_cookie(**cookie_kwargs)

pylti1p3/contrib/fastapi/launch_data_storage/cache.py

@@ -0,0 +1,9 @@
+from pylti1p3.launch_data_storage.cache import CacheDataStorage
+
+
+class FastAPICacheDataStorage(CacheDataStorage):
+    _cache = None
+
+    def __init__(self, cache, **kwargs):
+        self._cache = cache
+        super().__init__(cache, **kwargs)

pylti1p3/contrib/fastapi/message_launch.py

@@ -0,0 +1,33 @@
+from pylti1p3.message_launch import MessageLaunch
+
+from .cookie import FastAPICookieService
+from .session import FastAPISessionService
+
+
+class FastAPIMessageLaunch(MessageLaunch):
+    def __init__(
+        self,
+        request,
+        tool_config,
+        session_service=None,
+        cookie_service=None,
+        launch_data_storage=None,
+        requests_session=None,
+    ):
+        cookie_service = (
+            cookie_service if cookie_service else FastAPICookieService(request)
+        )
+        session_service = (
+            session_service if session_service else FastAPISessionService(request)
+        )
+        super().__init__(
+            request,
+            tool_config,
+            session_service,
+            cookie_service,
+            launch_data_storage,
+            requests_session,
+        )
+
+    def _get_request_param(self, key):
+        return self._request.get_param(key)

pylti1p3/contrib/fastapi/oidc_login.py

@@ -0,0 +1,37 @@
+from fastapi.responses import HTMLResponse
+
+from pylti1p3.oidc_login import OIDCLogin
+
+from .cookie import FastAPICookieService
+from .redirect import FastAPIRedirect
+from .session import FastAPISessionService
+
+
+class FastAPIOIDCLogin(OIDCLogin):
+    def __init__(
+        self,
+        request,
+        tool_config,
+        session_service=None,
+        cookie_service=None,
+        launch_data_storage=None,
+    ):
+        cookie_service = (
+            cookie_service if cookie_service else FastAPICookieService(request)
+        )
+        session_service = (
+            session_service if session_service else FastAPISessionService(request)
+        )
+        super().__init__(
+            request,
+            tool_config,
+            session_service,
+            cookie_service,
+            launch_data_storage,
+        )
+
+    def get_redirect(self, url):
+        return FastAPIRedirect(url, self._cookie_service)
+
+    def get_response(self, html):
+        return HTMLResponse(content=html)

pylti1p3/contrib/fastapi/redirect.py

@@ -0,0 +1,34 @@
+from fastapi.responses import HTMLResponse, RedirectResponse
+
+from pylti1p3.redirect import Redirect
+
+
+class FastAPIRedirect(Redirect):
+    _location = None
+    _cookie_service = None
+
+    def __init__(self, location, cookie_service=None):
+        super().__init__()
+        self._location = location
+        self._cookie_service = cookie_service
+
+    def do_redirect(self):
+        return self._process_response(RedirectResponse(self._location, status_code=302))
+
+    def do_js_redirect(self):
+        return self._process_response(
+            HTMLResponse(
+                f'<script type="text/javascript">window.location="{self._location}";</script>'
+            )
+        )
+
+    def set_redirect_url(self, location):
+        self._location = location
+
+    def get_redirect_url(self):
+        return self._location
+
+    def _process_response(self, response):
+        if self._cookie_service:
+            self._cookie_service.update_response(response)
+        return response

pylti1p3/contrib/fastapi/request.py

@@ -0,0 +1,35 @@
+from pylti1p3.request import Request
+
+
+class FastAPIRequest(Request):
+    _request = None
+    _form_data = None
+
+    def __init__(self, request, form_data):
+        """
+        Parameters:
+            request: FastAPI request
+            form_data: form data from FastAPI request
+                To get form data from FastAPI request, must use async method.
+                As we don't use async functions here, form data must be provided from outside.
+        """
+
+        super().__init__()
+
+        self._request = request
+        self._form_data = form_data
+
+    @property
+    def session(self):
+        return self._request.session
+
+    def get_param(self, key):
+        if self._request.method == "GET":
+            return self._request.query_params.get(key, None)
+        return self._form_data.get(key)
+
+    def get_cookie(self, key):
+        return self._request.cookies.get(key, None)
+
+    def is_secure(self):
+        return self._request.url.is_secure

pylti1p3/contrib/fastapi/session.py

@@ -0,0 +1,5 @@
+from pylti1p3.session import SessionService
+
+
+class FastAPISessionService(SessionService):
+    pass

pylti1p3/contrib/flask/cookie.py

@@ -20,14 +20,14 @@ def set_cookie(self, name, value, exp=3600):
 
     def update_response(self, response):
         for key, cookie_data in self._cookie_data_to_set.items():
-            cookie_kwargs = dict(
-                key=key,
-                value=cookie_data["value"],
-                max_age=cookie_data["exp"],
-                secure=self._request.is_secure(),
-                path="/",
-                httponly=True,
-            )
+            cookie_kwargs = {
+                "key": key,
+                "value": cookie_data["value"],
+                "max_age": cookie_data["exp"],
+                "secure": self._request.is_secure(),
+                "path": "/",
+                "httponly": True,
+            }
 
             if self._request.is_secure():
                 cookie_kwargs["samesite"] = "None"

pylti1p3/tool_config/abstract.py

@@ -39,14 +39,14 @@ def check_iss_has_many_clients(self, iss: str) -> bool:
         return iss_type == IssuerToClientRelation.MANY_CLIENTS_IDS_PER_ISSUER
 
     def set_iss_has_one_client(self, iss: str):
-        self.issuers_relation_types[
-            iss
-        ] = IssuerToClientRelation.ONE_CLIENT_ID_PER_ISSUER
+        self.issuers_relation_types[iss] = (
+            IssuerToClientRelation.ONE_CLIENT_ID_PER_ISSUER
+        )
 
     def set_iss_has_many_clients(self, iss: str):
-        self.issuers_relation_types[
-            iss
-        ] = IssuerToClientRelation.MANY_CLIENTS_IDS_PER_ISSUER
+        self.issuers_relation_types[iss] = (
+            IssuerToClientRelation.MANY_CLIENTS_IDS_PER_ISSUER
+        )
 
     def find_registration(self, iss: str, *args, **kwargs) -> Registration:
         """

tests/test_resource_link.py

@@ -293,9 +293,9 @@ def _get_data_with_invalid_deployment(
 
     def _get_data_with_invalid_message(self, *args):  # pylint: disable=unused-argument
         message_launch_data = self.expected_message_launch_data.copy()
-        message_launch_data[
-            "https://purl.imsglobal.org/spec/lti/claim/version"
-        ] = "1.2.0"
+        message_launch_data["https://purl.imsglobal.org/spec/lti/claim/version"] = (
+            "1.2.0"
+        )
         return message_launch_data
 
     def test_res_link_launch_invalid_nonce(self):

tox.ini

@@ -13,6 +13,7 @@ deps =
     black
     coverage
     django
+    fastapi
     flake8
     flask
     jwcrypto