fleshed out token auth

piccolo_api/token_auth/endpoints.py

@@ -0,0 +1,34 @@
+from starlette.endpoints import HTTPEndpoint
+from starlette.exceptions import HTTPException
+from starlette.responses import JSONResponse
+from starlette.requests import Request
+from piccolo.extensions.user.tables import BaseUser
+
+from .tables import TokenAuth
+
+
+class TokenAuthLoginEndpoint(HTTPEndpoint):
+    async def post(self, request: Request) -> JSONResponse:
+        """
+        Return a token if the credentials are correct.
+        """
+        json = await request.json()
+        username = json.get("username")
+        password = json.get("password")
+        if username and password:
+            user = await BaseUser.login(username=username, password=password)
+            if user:
+                token = (
+                    TokenAuth.select(TokenAuth.token)
+                    .first()
+                    .where(TokenAuth.user == user)
+                )
+                return JSONResponse({"token": token})
+            else:
+                raise HTTPException(
+                    status_code=401, detail="The credentials were incorrect"
+                )
+        else:
+            raise HTTPException(
+                status_code=401, detail="No credentials were found."
+            )

piccolo_api/token_auth/tables/token_auth.py → piccolo_api/token_auth/tables.py

@@ -1,4 +1,6 @@
-import secrets
+import uuid
+
+from asgiref.sync import async_to_sync
 
 from piccolo.columns.column_types import Varchar, ForeignKey
 from piccolo.extensions.user.tables import BaseUser
@@ -12,9 +14,15 @@ class TokenAuth(Table):
     Useful for mobile authentication, IOT etc. Session auth is recommended for
     web usage.
     """
-    token = Varchar()
+
+    token = Varchar(default=uuid.uuid4)
     user = ForeignKey(references=BaseUser)
 
     @classmethod
-    def create_token(user_id: int):
-        pass
+    async def create_token(cls, user_id: int) -> str:
+        token_auth = await cls(user=user_id).save().run()
+        return token_auth.token
+
+    @classmethod
+    def create_token_sync(cls, user_id: int) -> str:
+        return async_to_sync(cls.create_token)(user_id)