__magic__

pyproject.toml

@@ -5,7 +5,7 @@
 
 [tool.poetry]
     name = "royalnet"
-    version = "5.9.3"
+    version = "5.10.0"
     description = "A multipurpose bot and web framework"
     authors = ["Stefano Pigozzi <ste.pigozzi@gmail.com>"]
     license = "AGPL-3.0+"

royalnet/backpack/stars/api_auth_login_royalnet.py

@@ -1,16 +1,14 @@
 import datetime
 import royalnet.utils as ru
-from royalnet.constellation.api import *
-from royalnet.constellation.api.apierrors import *
+import royalnet.constellation.api as rca
+import royalnet.constellation.api.apierrors as rcae
 from ..tables.users import User
 from ..tables.tokens import Token
 
 
-class ApiAuthLoginRoyalnetStar(ApiStar):
+class ApiAuthLoginRoyalnetStar(rca.ApiStar):
     path = "/api/auth/login/royalnet/v1"
 
-    methods = ["POST"]
-
     parameters = {
         "post": {
             "username": "The name of the user you are logging in as.",
@@ -20,7 +18,8 @@ class ApiAuthLoginRoyalnetStar(ApiStar):
 
     tags = ["auth"]
 
-    async def post(self, data: ApiData) -> ru.JSON:
+    @rca.magic
+    async def post(self, data: rca.ApiData) -> ru.JSON:
         """Login with a Royalnet account.
 
         The method returns a API token valid for 7 days that identifies and authenticates the user to the API.
@@ -36,10 +35,10 @@ class ApiAuthLoginRoyalnetStar(ApiStar):
         async with self.session_acm() as session:
             user: User = await ru.asyncify(session.query(UserT).filter_by(username=username).one_or_none)
             if user is None:
-                raise NotFoundError("User not found")
+                raise rcae.NotFoundError("User not found")
             pswd_check = user.test_password(password)
             if not pswd_check:
-                raise UnauthorizedError("Invalid password")
+                raise rcae.UnauthorizedError("Invalid password")
             token: Token = TokenT.generate(alchemy=self.alchemy, user=user, expiration_delta=datetime.timedelta(days=7))
             session.add(token)
             await ru.asyncify(session.commit)

royalnet/backpack/stars/api_auth_token.py

@@ -1,15 +1,13 @@
 from typing import *
 import datetime
 import royalnet.utils as ru
-from royalnet.constellation.api import *
+import royalnet.constellation.api as rca
 from ..tables.tokens import Token
 
 
-class ApiAuthTokenStar(ApiStar):
+class ApiAuthTokenStar(rca.ApiStar):
     path = "/api/auth/token/v1"
 
-    methods = ["GET", "POST"]
-
     parameters = {
         "get": {},
         "post": {
@@ -24,12 +22,14 @@ class ApiAuthTokenStar(ApiStar):
 
     tags = ["auth"]
 
-    async def get(self, data: ApiData) -> ru.JSON:
+    @rca.magic
+    async def get(self, data: rca.ApiData) -> ru.JSON:
         """Get information about the current login token."""
         token = await data.token()
         return token.json()
 
-    async def post(self, data: ApiData) -> ru.JSON:
+    @rca.magic
+    async def post(self, data: rca.ApiData) -> ru.JSON:
         """Create a new login token for the authenticated user.
 
         Keep it secret, as it is basically a password!"""
@@ -37,7 +37,7 @@ class ApiAuthTokenStar(ApiStar):
         try:
             duration = int(data["duration"])
         except ValueError:
-            raise InvalidParameterError("Duration is not a valid integer")
+            raise rca.InvalidParameterError("Duration is not a valid integer")
         new_token = Token.generate(self.alchemy, user, datetime.timedelta(seconds=duration))
         data.session.add(new_token)
         await data.session_commit()

royalnet/backpack/stars/api_royalnet_version.py

@@ -1,16 +1,15 @@
 import royalnet.version as rv
-from royalnet.constellation.api import *
+import royalnet.constellation.api as rca
 import royalnet.utils as ru
 
 
-class ApiRoyalnetVersionStar(ApiStar):
+class ApiRoyalnetVersionStar(rca.ApiStar):
     path = "/api/royalnet/version/v1"
 
-    methods = ["GET"]
-
     tags = ["royalnet"]
 
-    async def get(self, data: ApiData) -> ru.JSON:
+    @rca.magic
+    async def get(self, data: rca.ApiData) -> ru.JSON:
         """Get the current Royalnet version."""
         return {
             "semantic": rv.semantic

royalnet/backpack/stars/api_user_create.py

@@ -9,8 +9,6 @@ from ..tables import *
 class ApiUserCreateStar(rca.ApiStar):
     path = "/api/user/create/v1"
 
-    methods = ["POST"]
-
     parameters = {
         "post": {
             "username": "The name of the user you are creating.",

royalnet/backpack/stars/api_user_find.py

@@ -1,22 +1,20 @@
-from royalnet.utils import *
-from royalnet.backpack.tables import *
-from royalnet.constellation.api import *
+import royalnet.utils as ru
+import royalnet.backpack.tables as rbt
+import royalnet.constellation.api as rca
 
 
-class ApiUserFindStar(ApiStar):
+class ApiUserFindStar(rca.ApiStar):
     path = "/api/user/find/v1"
 
-    tags = ["user"]
-
     parameters = {
         "get": {
             "alias": "One of the aliases of the user to get."
         }
     }
 
-    async def get(self, data: ApiData) -> dict:
+    async def get(self, data: rca.ApiData) -> ru.JSON:
         """Get details about the Royalnet user with a certain alias."""
-        user = await User.find(self.alchemy, data.session, data["alias"])
+        user = await rbt.User.find(self.alchemy, data.session, data["alias"])
         if user is None:
-            raise NotFoundError("No such user.")
+            raise rca.NotFoundError("No such user.")
         return user.json()

royalnet/backpack/stars/api_user_get.py

@@ -1,9 +1,9 @@
-from royalnet.utils import *
-from royalnet.backpack.tables import *
-from royalnet.constellation.api import *
+import royalnet.utils as ru
+import royalnet.backpack.tables as rbt
+import royalnet.constellation.api as rca
 
 
-class ApiUserGetStar(ApiStar):
+class ApiUserGetStar(rca.ApiStar):
     path = "/api/user/get/v1"
 
     parameters = {
@@ -14,14 +14,15 @@ class ApiUserGetStar(ApiStar):
 
     tags = ["user"]
 
-    async def get(self, data: ApiData) -> dict:
+    @rca.magic
+    async def get(self, data: rca.ApiData) -> dict:
         """Get details about the Royalnet user with a certain id."""
         user_id_str = data["id"]
         try:
             user_id = int(user_id_str)
         except (ValueError, TypeError):
-            raise InvalidParameterError("'id' is not a valid int.")
-        user: User = await asyncify(data.session.query(self.alchemy.get(User)).get, user_id)
+            raise rca.InvalidParameterError("'id' is not a valid int.")
+        user: rbt.User = await ru.asyncify(data.session.query(self.alchemy.get(rbt.User)).get, user_id)
         if user is None:
-            raise NotFoundError("No such user.")
+            raise rca.NotFoundError("No such user.")
         return user.json()

royalnet/backpack/stars/api_user_list.py

@@ -1,15 +1,16 @@
 from starlette.responses import *
-from royalnet.utils import *
-from royalnet.backpack.tables import *
-from royalnet.constellation.api import *
+import royalnet.utils as ru
+import royalnet.backpack.tables as rbt
+import royalnet.constellation.api as rca
 
 
-class ApiUserListStar(ApiStar):
+class ApiUserListStar(rca.ApiStar):
     path = "/api/user/list/v1"
 
     tags = ["user"]
 
-    async def get(self, data: ApiData) -> JSON:
-        "Get a list of all Royalnet users."
-        users: typing.List[User] = await asyncify(data.session.query(self.alchemy.get(User)).all)
+    @rca.magic
+    async def get(self, data: rca.ApiData) -> ru.JSON:
+        """Get a list of all Royalnet users."""
+        users: typing.List[rbt.User] = await ru.asyncify(data.session.query(self.alchemy.get(rbt.User)).all)
         return [user.json() for user in users]

royalnet/backpack/stars/api_user_passwd.py

@@ -1,16 +1,14 @@
 from typing import *
 import datetime
 import royalnet.utils as ru
-from royalnet.constellation.api import *
+import royalnet.constellation.api as rca
 from sqlalchemy import and_
 from ..tables.tokens import Token
 
 
-class ApiUserPasswd(ApiStar):
+class ApiUserPasswd(rca.ApiStar):
     path = "/api/user/passwd/v1"
 
-    methods = ["PUT"]
-
     tags = ["user"]
 
     parameters = {
@@ -25,7 +23,8 @@ class ApiUserPasswd(ApiStar):
 
     requires_auth = True
 
-    async def put(self, data: ApiData) -> ru.JSON:
+    @rca.magic
+    async def put(self, data: rca.ApiData) -> ru.JSON:
         """Change the password of the currently logged in user.
 
         This method also revokes all the issued tokens for the user."""

royalnet/constellation/api/__init__.py

@@ -11,6 +11,7 @@ from .apierrors import \
     InvalidParameterError, \
     MethodNotImplementedError, \
     UnsupportedError
+from .magic import magic
 
 
 __all__ = [
@@ -31,4 +32,5 @@ __all__ = [
     "InvalidParameterError",
     "MethodNotImplementedError",
     "UnsupportedError",
+    "magic",
 ]

royalnet/constellation/api/apistar.py

@@ -20,6 +20,7 @@ class ApiStar(PageStar, ABC):
     deprecated: Dict[str, bool] = {}
 
     tags: List[str] = []
+    __override__: List[str] = []
 
     async def page(self, request: Request) -> JSONResponse:
         if request.query_params:
@@ -42,23 +43,25 @@ class ApiStar(PageStar, ABC):
                 response = await self.put(apidata)
             elif method == "delete":
                 response = await self.delete(apidata)
+            elif method == "options":
+                return api_success("Preflight allowed.", methods=self.methods())
             else:
                 raise MethodNotImplementedError("Unknown method")
         except UnauthorizedError as e:
-            return api_error(e, code=401)
+            return api_error(e, code=401, methods=self.methods())
         except NotFoundError as e:
-            return api_error(e, code=404)
+            return api_error(e, code=404, methods=self.methods())
         except ForbiddenError as e:
-            return api_error(e, code=403)
+            return api_error(e, code=403, methods=self.methods())
         except MethodNotImplementedError as e:
-            return api_error(e, code=405)
+            return api_error(e, code=405, methods=self.methods())
         except BadRequestError as e:
-            return api_error(e, code=400)
+            return api_error(e, code=400, methods=self.methods())
         except Exception as e:
             ru.sentry_exc(e)
-            return api_error(e, code=500)
+            return api_error(e, code=500, methods=self.methods())
         else:
-            return api_success(response)
+            return api_success(response, methods=self.methods())
         finally:
             await apidata.session_close()
 
@@ -107,10 +110,23 @@ class ApiStar(PageStar, ABC):
             }
         }
 
+    @classmethod
+    def methods(cls):
+        magics = []
+        for key, value in cls.__dict__.items():
+            attr = value
+            try:
+                magic = attr.__magic__
+            except AttributeError:
+                continue
+            if magic:
+                magics.append(key)
+        return [*magics, "options"]
+
     def swagger(self) -> ru.JSON:
         """Generate one or more swagger paths for this ApiStar."""
         result = {}
-        for method in self.methods:
+        for method in self.methods():
             result[method.lower()] = self.__swagger_for_a_method(self.__getattribute__(method.lower()))
         return result
 

royalnet/constellation/api/jsonapi.py

@@ -1,33 +1,38 @@
 from typing import *
+import royalnet.utils as ru
 try:
     from starlette.responses import JSONResponse
 except ImportError:
     JSONResponse = None
 
 
-def api_response(data: dict, code: int, headers: dict = None) -> JSONResponse:
+def api_response(data: ru.JSON, code: int, headers: dict = None, methods=None) -> JSONResponse:
     if headers is None:
         headers = {}
+    if methods is None:
+        methods = ["GET"]
+
     full_headers = {
         **headers,
         "Access-Control-Allow-Origin": "*",
+        "Access-Control-Allow-Methods": ", ".join(methods).upper()
     }
     return JSONResponse(data, status_code=code, headers=full_headers)
 
 
-def api_success(data: dict) -> JSONResponse:
+def api_success(data: ru.JSON, methods=None) -> JSONResponse:
     result = {
         "success": True,
         "data": data
     }
-    return api_response(result, code=200)
+    return api_response(result, code=200, methods=methods)
 
 
-def api_error(error: Exception, code: int = 500) -> JSONResponse:
+def api_error(error: Exception, code: int = 500, methods=None) -> JSONResponse:
     result = {
         "success": False,
         "error_type": error.__class__.__qualname__,
         "error_args": list(error.args),
         "error_code": code,
     }
-    return api_response(result, code=code)
+    return api_response(result, code=code, methods=methods)

royalnet/constellation/api/magic.py

@@ -0,0 +1,15 @@
+from typing import *
+import functools
+
+
+def magic(func):
+    # i made this at 2:00 am
+    # at 2:15 am i already had no idea on how it worked
+    # at 2:30 i still have no idea but it works appearently
+    # 2:40 TODO: document me
+    func.__magic__ = True
+
+    @functools.wraps(func)
+    async def f(*args, **kwargs):
+        return await func(*args, **kwargs)
+    return f

royalnet/constellation/constellation.py

@@ -98,16 +98,7 @@ class Constellation:
         self.events: Dict[str, rc.Event] = {}
         """A dictionary containing all :class:`~rc.Event` that can be handled by this :class:`Constellation`."""
 
-        middleware = []
-        if constellation_cfg.get("cors_middleware", False):
-            log.info("CORS middleware: enabled")
-            middleware.append(
-                starlette.middleware.Middleware(starlette.middleware.cors.CORSMiddleware, {"allow_origins": ["*"]})
-            )
-        else:
-            log.info("CORS middleware: disabled")
-
-        self.starlette = starlette.applications.Starlette(debug=__debug__, middleware=middleware)
+        self.starlette = starlette.applications.Starlette(debug=__debug__)
         """The :class:`~starlette.Starlette` app."""
 
         self.stars: List[PageStar] = []
@@ -269,7 +260,7 @@ class Constellation:
             log.info(f"Running {page_star}")
             return await page_star.page(request)
 
-        return page_star.path, f, page_star.methods
+        return page_star.path, f, page_star.methods()
 
     def register_page_stars(self, page_stars: List[Type[PageStar]], pack_cfg: Dict[str, Any]):
         for SelectedPageStar in page_stars:

royalnet/constellation/pagestar.py

@@ -18,7 +18,8 @@ class PageStar(Star):
 
     """
 
-    methods: List[str] = ["GET"]
+    @classmethod
+    def methods(cls):
         """The HTTP methods supported by the Star, in form of a list.
 
         By default, a Star only supports the ``GET`` method, but more can be added.
@@ -29,6 +30,7 @@ class PageStar(Star):
                 methods: List[str] = ["GET", "POST", "PUT", "DELETE"]
 
         """
+        return ["GET"]
 
     def __repr__(self):
         return f"<{self.__class__.__qualname__}: {self.path}>"

royalnet/version.py

@@ -1 +1 @@
-semantic = "5.9.3"
+semantic = "5.10.0"