add backend folder

Author: sinisaos

.gitignore renamed to backend/.gitignore

@@ -0,0 +1,21 @@
+# piccolo_project
+
+## Setup
+
+### Install requirements
+
+```bash
+pip install -r requirements.txt
+```
+
+### Getting started guide
+
+```bash
+python main.py
+```
+
+### Running tests
+
+```bash
+piccolo tester run
+```

backend/README.md

@@ -0,0 +1,116 @@
+from blacksheep import FromJSON, json
+from blacksheep.cookies import Cookie
+from blacksheep.server.controllers import Controller, get, post
+from blacksheep.server.responses import redirect
+from piccolo.apps.user.tables import BaseUser
+from piccolo_api.session_auth.tables import SessionsBase
+
+from accounts.schema import UserModelLogin, UserModelRegister
+
+
+class AuthController(Controller):
+    @post("/register")
+    async def register(request, data: FromJSON[UserModelRegister]):
+        """
+        Register and authenticate user
+        """
+        data = data.__dict__
+        payload = data["_value"].dict()
+        if (
+            await BaseUser.exists().where(BaseUser.email == payload["email"]).run()
+            or await BaseUser.exists()
+            .where(BaseUser.username == payload["username"])
+            .run()
+        ):
+            user_error = "User with that email or username already exists."
+            return json({"error": user_error})
+        # save user
+        query = BaseUser(**payload)
+        await query.save().run()
+        # login user in
+        valid_user = await BaseUser.login(
+            username=payload["username"], password=payload["password"]
+        )
+        if not valid_user:
+            user_error = "Invalid username or password"
+            return json({"error": user_error})
+        # create session
+        session = await SessionsBase.create_session(user_id=valid_user)
+        response = redirect("/")
+        response.set_cookie(
+            Cookie(
+                "id",
+                f"{session['token']}",
+            )
+        )
+        response.add_header(b"Access-Control-Allow-Credentials", b"true")
+        return response
+
+    @post("/login")
+    async def login(request, data: FromJSON[UserModelLogin]):
+        """
+        Login and authenticate user
+        """
+        data = data.__dict__
+        payload = data["_value"].dict()
+        # login user in
+        valid_user = await BaseUser.login(
+            username=payload["username"], password=payload["password"]
+        )
+        if not valid_user:
+            user_error = "Invalid username or password"
+            return json({"error": user_error})
+        # create session
+        session = await SessionsBase.create_session(user_id=valid_user)
+        response = redirect("/profile")
+        response.set_cookie(
+            Cookie(
+                "id",
+                f"{session['token']}",
+                http_only=True,
+            )
+        )
+        response.add_header(b"Access-Control-Allow-Credentials", b"true")
+        return response
+
+    @get("/profile")
+    async def profile(request):
+        """
+        User profile
+        """
+        data = (
+            await SessionsBase.select(SessionsBase.user_id)
+            .where(SessionsBase.token == request.cookies.get("id"))
+            .first()
+            .run()
+        )
+        if data:
+            session_user = (
+                await BaseUser.select(
+                    BaseUser.id,
+                    BaseUser.username,
+                    BaseUser.email,
+                    BaseUser.last_login,
+                )
+                .where(BaseUser._meta.primary_key == data["user_id"])
+                .first()
+                .run()
+            )
+        user = session_user
+        response = json(
+            {
+                "user": user,
+            }
+        )
+        response.add_header(b"Access-Control-Allow-Credentials", b"true")
+        return response
+
+    @post("/logout")
+    async def logout(request):
+        """
+        Logout user
+        """
+        response = redirect("/")
+        response.unset_cookie("id")
+        response.add_header(b"Access-Control-Allow-Credentials", b"true")
+        return response

backend/accounts/__init__.py

@@ -0,0 +1,21 @@
+from piccolo.apps.user.tables import BaseUser
+from piccolo_api.crud.serializers import create_pydantic_model
+
+# user models
+UserModelRegister = create_pydantic_model(
+    table=BaseUser,
+    exclude_columns=(
+        BaseUser.active,
+        BaseUser.admin,
+        BaseUser.superuser,
+    ),
+    model_name="UserModelRegister",
+)
+UserModelLogin = create_pydantic_model(
+    table=BaseUser,
+    include_columns=(
+        BaseUser.username,
+        BaseUser.password,
+    ),
+    model_name="UserModelLogin",
+)

backend/accounts/endpoints.py

@@ -0,0 +1,88 @@
+from blacksheep import Application, json
+from blacksheep.server.openapi.v3 import OpenAPIHandler
+from openapidocs.v3 import Info
+from piccolo.apps.user.tables import BaseUser
+from piccolo.engine import engine_finder
+from piccolo_admin.endpoints import create_admin
+from piccolo_api.session_auth.tables import SessionsBase
+
+from accounts.endpoints import AuthController
+from home.endpoints import Tasks
+from home.tables import Task
+
+app = Application(show_error_details=True)
+
+
+app.use_cors(
+    allow_origins="http://localhost:8080",
+    allow_methods="*",
+    allow_headers="*",
+    allow_credentials=True,
+    expose_headers="next_cursor,first_row,last_row",
+)
+
+# mount Piccolo ASGI apps
+app.mount(
+    "/admin/",
+    create_admin(
+        tables=[Task, BaseUser, SessionsBase],
+        # Required when running under HTTPS:
+        # allowed_hosts=['my_site.com']
+    ),
+)
+
+# Openpi docs
+docs = OpenAPIHandler(info=Info(title="BlackSheep API", version="0.0.1"))
+docs.bind_app(app)
+
+
+@docs.ignore()
+async def home(request) -> json:
+    try:
+        data = (
+            await SessionsBase.select(SessionsBase.user_id)
+            .where(SessionsBase.token == request.cookies.get("id"))
+            .first()
+            .run()
+        )
+        if data:
+            session_user = (
+                await BaseUser.select(BaseUser.username)
+                .where(BaseUser._meta.primary_key == data["user_id"])
+                .first()
+                .run()
+            )
+        auth_user = session_user["username"]
+    except UnboundLocalError:
+        auth_user = ""
+    response = json(
+        {
+            "message": "BlackSheep Piccolo ORM api",
+            "auth_user": auth_user,
+        }
+    )
+    response.add_header(b"Access-Control-Allow-Credentials", b"true")
+    return response
+
+
+app.router.add_get("/", home)
+
+
+async def open_database_connection_pool(application):
+    try:
+        engine = engine_finder()
+        await engine.start_connection_pool()
+    except Exception:
+        print("Unable to connect to the database")
+
+
+async def close_database_connection_pool(application):
+    try:
+        engine = engine_finder()
+        await engine.close_connection_pool()
+    except Exception:
+        print("Unable to connect to the database")
+
+
+app.on_start += open_database_connection_pool
+app.on_stop += close_database_connection_pool

backend/accounts/schema.py

@@ -0,0 +1,17 @@
+import os
+import sys
+
+from piccolo.utils.warnings import colored_warning
+
+
+def pytest_configure(*args):
+    if os.environ.get("PICCOLO_TEST_RUNNER") != "True":
+        colored_warning(
+            "\n\n"
+            "We recommend running Piccolo tests using the "
+            "`piccolo tester run` command, which wraps Pytest, and makes "
+            "sure the test database is being used. "
+            "To stop this warning, modify conftest.py."
+            "\n\n"
+        )
+        sys.exit(1)