Mealie beta

.gitignore

@@ -6,6 +6,9 @@ __pycache__/
 # C extensions
 *.so
 
+# Ruff
+.ruff_cache/
+
 # Pycharm
 .idea/
 venv/

Dockerfile

@@ -0,0 +1,4 @@
+ARG BASE_IMAGE=python:latest
+FROM ${BASE_IMAGE} AS base
+
+ENTRYPOINT [ "echo", "Hello World!" ]

compose.yml

@@ -0,0 +1,41 @@
+version: "3.9"
+
+services:
+  tests:
+    build:
+      context: .
+      args:
+        BASE_IMAGE: "python:3.9"
+    image: mealieapi-tests:latest
+    volumes:
+      - ./volumes/coverage:/app/coverage:rw
+    depends_on:
+      - mealie-api
+    environment:
+      MEALIE_URL: http://mealie-api:9000
+      MEALIE_TOKEN: ${MEALIE_TEST_SUITE_TOKEN}
+    restart: unless-stopped
+  mealie-frontend:
+    image: hkotel/mealie:frontend-v1.0.0beta-5
+    container_name: mealie-frontend
+    environment:
+      - API_URL=http://mealie-api:9000
+    restart: always
+    ports:
+      - "9925:3000"
+    volumes:
+      - mealie-data:/app/data/
+  mealie-api:
+    image: hkotel/mealie:api-v1.0.0beta-5
+      - API_PORT=9000
+      - DEFAULT_GROUP="Testers"
+      - DEFAULT_EMAIL="tester@email.com"
+      - DEFAULT_PASSWORD="agloriouspasswordfortesting"
+    restart: unless-stopped
+      - DEFAULT_EMAIL=tester101@email.com
+      - DEFAULT_GROUP=mealieapi
+    restart: always
+
+volumes:
+  mealie-data:
+    driver: local

mealieapi/__init__.py

@@ -1,3 +1 @@
-from mealieapi.client import MealieClient
-
-__version__ = "0.0.0"
+from .client import Client

mealieapi/auth.py

@@ -1,35 +1,122 @@
-from __future__ import annotations
+from posixpath import join as urljoin
+from typing import Any, cast
 
-import typing as t
+from aiohttp import ClientSession
 
-import aiohttp
 
-from mealieapi.model import InteractiveModel
+class Auth:
+    REFRESH_AFTER: int = 60 * 60 * 12
 
+    def __init__(
+        self,
+        url: str,
+        access_token: str | None = None,
+        token_type: str | None = None,
+    ) -> None:
+        self._api = url
+        self._access_token: str | None = access_token
+        self._token_type: str | None = token_type
+        self._session = ClientSession(headers=self.session_headers)
 
-class Token(InteractiveModel):
-    name: str
-    id: int
-    token: str | None = None
+    async def __aenter__(self) -> "Auth":
+        return self
 
-    def dict(self, *args, **kwargs) -> dict[str, t.Any]:  # type: ignore[override]
-        data = super().dict(*args, **kwargs)
-        data.pop("token")
-        return data
+    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+        await self.close()
 
-    async def delete(self) -> None:
-        await self._client.delete_api_key(self.id)
+    async def close(self) -> None:
+        await self._session.close()
 
+    async def _headers(self) -> dict[str, str]:
+        print(self._access_token)
+        if self._access_token is not None:
+            return {
+                "Authorization": f"{self._token_type.capitalize()} {self._access_token}"
+            }
+        return {}
 
-class Auth(InteractiveModel):
-    access_token: str
-    token_type: str | None = None
+    async def request(
+        self,
+        method: str,
+        endpoint: str,
+        headers: dict[str, str] | None = None,
+        use_auth: bool = True,
+        params: dict[str, str] | None = None,
+        **kwargs: Any,
+    ) -> dict[str, Any] | bytes:
+        """Make a request to the API with proper authentication."""
+        _headers = {} if not use_auth else await self._headers()
+        response = await self._session.request(
+            method,
+            urljoin(self._api, endpoint),
+            headers=_headers | (headers or {}),
+            params=params,
+            **kwargs,
+        )
+        # TODO: Add mimetype hooks and json model processing
+        if response.content_type == "application/json":
+            return await response.json()
+        if response.content_type == "text/html":
+            raise UserWarning(
+                "Did you request the wrong endpoint? The response was HTML."
+            )
+        if response.content_type.startswith("text/"):
+            return await response.text()
+        return await response.content.read()
 
-    async def refresh(self):
-        resp = await self._client.request("auth/refresh")
-        self.access_token = resp["access_token"]
-        self.token_type = resp["token_type"]
+    async def get_token(
+        self,
+        username: str,
+        password: str,
+        grant_type: str | None = None,
+        scope: str | None = None,
+        client_id: str | None = None,
+        client_secret: str | None = None,
+        long_term: bool = False,
+    ) -> dict[str, str]:
+        """Get a session token from the API."""
+        data = await self.request(
+            method="POST",
+            endpoint="api/auth/token",
+            data={
+                "username": username,
+                "password": password,
+                "grant_type": grant_type if grant_type is not None else "password",
+                "scope": scope if scope is not None else "",
+                "client_id": client_id if client_id is not None else "",
+                "client_secret": client_secret if client_secret is not None else "",
+                "remember_me": "true" if long_term else "false",
+            },
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+            use_auth=False,
+        )
+        return cast(dict[str, str], data)
 
-    @property
-    def header(self):
-        return {aiohttp.hdrs.AUTHORIZATION: f"Bearer {self.access_token}"}
+    async def refresh_token(self) -> dict[str, str]:
+        """Refresh the session token."""
+        data = await self.request(
+            method="GET",
+            endpoint="api/auth/refresh",
+        )
+        return cast(dict[str, str], data)
+
+
+class AccessTokenAuth(Auth):
+    def __init__(
+        self, url: str, access_token: str | None, token_type: str = "bearer"
+    ) -> None:
+        super().__init__(url, access_token, token_type)
+
+
+class BasicAuth(Auth):
+    def __init__(self, url: str, username: str, password: str) -> None:
+        self._username = username
+        self._password = password
+        super().__init__(url, None)
+
+    async def _headers(self) -> dict[str, str]:
+        if self._access_token is None:
+            data = await self.get_token(self._username, self._password)
+            self._access_token = data.get("access_token")
+            self._token_type = data.get("token_type")
+        return await super()._headers()