Mealie Token Authentication

.flake8

@@ -2,3 +2,4 @@
 ignore = D203,W503
 exclude = .git,__pycache__,build,dist
 max-line-length = 119
+extend-ignore = E701

kptncook/__init__.py

@@ -9,12 +9,14 @@
 
 import httpx
 import typer
+from pydantic import ValidationError
 from rich import print as rprint
 from rich.pretty import pprint
 from rich.prompt import Prompt
 
+from kptncook.config import Settings
+
 from .api import KptnCookClient, parse_id
-from .config import settings
 from .mealie import MealieApiClient, kptncook_to_mealie
 from .models import Recipe
 from .paprika import PaprikaExporter
@@ -33,15 +35,34 @@
 ]
 
 __version__ = "0.0.21"
-cli = typer.Typer()
+
+
+def load_settings():
+    global settings
+    try:
+        settings = Settings()  # type: ignore
+    except ValidationError as e:
+        rprint("validation error: ", e)
+        sys.exit(1)
+
+
+def create_kptncook_client():
+    return KptnCookClient(
+        base_url=settings.kptncook_api_url,
+        api_key=settings.kptncook_api_key,
+        access_token=settings.kptncook_access_token,
+    )
+
+
+cli = typer.Typer(callback=load_settings)
 
 
 @cli.command(name="kptncook-today")
 def list_kptncook_today():
     """
     List all recipes for today from the kptncook site.
     """
-    client = KptnCookClient()
+    client = create_kptncook_client()
     all_recipes = client.list_today()
     for recipe in all_recipes:
         pprint(recipe)
@@ -54,13 +75,17 @@ def save_todays_recipes():
     """
     fs_repo = RecipeRepository(settings.root)
     if fs_repo.needs_to_be_synced(date.today()):
-        client = KptnCookClient()
+        client = create_kptncook_client()
         fs_repo.add_list(client.list_today())
 
 
 def get_mealie_client() -> MealieApiClient:
     client = MealieApiClient(settings.mealie_url)
-    client.login(settings.mealie_username, settings.mealie_password)
+
+    if settings.mealie_api_token:
+        client.login_with_token(settings.mealie_api_token)
+    else:
+        client.login(settings.mealie_username, settings.mealie_password)
     return client
 
 
@@ -114,7 +139,9 @@ def sync_with_mealie():
         sys.exit(1)
     kptncook_recipes_from_mealie = get_kptncook_recipes_from_mealie(client)
     recipes = get_kptncook_recipes_from_repository()
-    kptncook_recipes_from_repository = [kptncook_to_mealie(r) for r in recipes]
+    kptncook_recipes_from_repository = [
+        kptncook_to_mealie(r, settings.kptncook_api_key) for r in recipes
+    ]
     ids_in_mealie = {r.extras["kptncook_id"] for r in kptncook_recipes_from_mealie}
     ids_from_api = {r.extras["kptncook_id"] for r in kptncook_recipes_from_repository}
     ids_to_add = ids_from_api - ids_in_mealie
@@ -154,7 +181,7 @@ def backup_kptncook_favorites():
     if settings.kptncook_access_token is None:
         rprint("Please set KPTNCOOK_ACCESS_TOKEN in your environment or .env file")
         sys.exit(1)
-    client = KptnCookClient()
+    client = create_kptncook_client()
     favorites = client.list_favorites()
     rprint(f"Found {len(favorites)} favorites")
     ids = [("oid", oid["identifier"]) for oid in favorites]
@@ -175,7 +202,7 @@ def get_kptncook_access_token():
     """
     username = Prompt.ask("Enter your kptncook email address")
     password = Prompt.ask("Enter your kptncook password", password=True)
-    client = KptnCookClient()
+    client = create_kptncook_client()
     access_token = client.get_access_token(username, password)
     rprint("your access token: ", access_token)
 
@@ -210,7 +237,7 @@ def search_kptncook_recipe_by_id(id_: str):
         sys.exit(1)
     id_type, id_value = parsed
     rprint(id_type, id_value)
-    client = KptnCookClient()
+    client = create_kptncook_client()
     recipes = client.get_by_ids([(id_type, id_value)])
     if len(recipes) == 0:
         rprint("Could not find recipe")
@@ -237,7 +264,7 @@ def export_recipes_to_paprika(_id: OptionalId = typer.Argument(None)):
         recipes = get_recipe_by_id(_id)
     else:
         recipes = get_kptncook_recipes_from_repository()
-    exporter = PaprikaExporter()
+    exporter = PaprikaExporter(settings.kptncook_api_key)
     filename = exporter.export(recipes=recipes)
     rprint(
         "\n The data was exported to '%s'. Open the export file with the Paprika App.\n"

kptncook/api.py

@@ -6,7 +6,6 @@
 
 import httpx
 
-from .config import settings
 from .repositories import RecipeInDb
 
 
@@ -29,9 +28,7 @@ class KptnCookClient:
     Client for the kptncook api.
     """
 
-    def __init__(
-        self, base_url=settings.kptncook_api_url, api_key=settings.kptncook_api_key
-    ):
+    def __init__(self, base_url, api_key, access_token=None):
         self.base_url = str(base_url)
         self.headers = {
             "content-type": "application/json",
@@ -40,8 +37,8 @@ def __init__(
             "hasIngredients": "yes",
         }
         self.api_key = api_key
-        if settings.kptncook_access_token is not None:
-            self.headers["Token"] = settings.kptncook_access_token
+        if access_token is not None:
+            self.headers["Token"] = access_token
 
     @property
     def logged_in(self):

kptncook/config.py

@@ -1,12 +1,11 @@
 """
 Base settings for kptncook.
 """
-import sys
+
 from pathlib import Path
 
-from pydantic import AnyHttpUrl, DirectoryPath, Field, ValidationError, field_validator
+from pydantic import AnyHttpUrl, DirectoryPath, Field, field_validator, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
-from rich import print as rprint
 
 
 class Settings(BaseSettings):
@@ -18,17 +17,19 @@ class Settings(BaseSettings):
     kptncook_access_token: str | None = None
     kptncook_api_url: AnyHttpUrl = AnyHttpUrl("https://mobile.kptncook.com")
     mealie_url: AnyHttpUrl = AnyHttpUrl("http://localhost:9000/api")
-    mealie_username: str
-    mealie_password: str
+
+    mealie_username: str | None = None
+    mealie_password: str | None = None
+    mealie_api_token: str | None = None
 
     @field_validator("root", mode="before")
     def root_must_exist(cls, path: Path) -> Path:
         path.mkdir(parents=True, exist_ok=True)
         return path
 
+    @model_validator(mode="after")
+    def check_mealie_auth(self):
+        if (self.mealie_password is None or self.mealie_username is None) and self.mealie_api_token is None:
+            raise ValueError("must specify either mealie_username/password or mealie_api_token")
 
-try:
-    settings = Settings()  # type: ignore
-except ValidationError as e:
-    rprint("validation error: ", e)
-    sys.exit(1)
+        return self

kptncook/mealie.py

@@ -9,7 +9,6 @@
 import httpx
 from pydantic import UUID4, BaseModel, Field, ValidationError, parse_obj_as
 
-from .config import settings
 from .models import Image
 from .models import Recipe as KptnCookRecipe
 
@@ -52,8 +51,7 @@ class UnitFoodBase(NameIsIdModel):
     description: str = ""
 
 
-class RecipeFood(UnitFoodBase):
-    ...
+class RecipeFood(UnitFoodBase): ...
 
 
 class RecipeUnit(UnitFoodBase):
@@ -189,11 +187,17 @@ def fetch_api_token(self, username, password):
         r.raise_for_status()
         return r.json()["access_token"]
 
+    def _set_token_header(self, access_token: str):
+        self.headers = {"authorization": f"Bearer {access_token}"}
+
     def login(self, username: str = "admin", password: str = ""):
         if password == "":
             password = getpass()
         access_token = self.fetch_api_token(username, password)
-        self.headers = {"authorization": f"Bearer {access_token}"}
+        self._set_token_header(access_token)
+
+    def login_with_token(self, token: str):
+        self._set_token_header(token)
 
     def upload_asset(self, recipe_slug, image: Image):
         # download image
@@ -230,7 +234,13 @@ def enrich_recipe_with_step_images(self, recipe):
             instruction.text = self._build_recipestep_text(
                 recipe.id, instruction.text, uploaded_image_name
             )
-            assets.append(RecipeAsset(name=asset_properties["name"], icon=asset_properties["icon"], file_name=asset_properties["fileName"]))
+            assets.append(
+                RecipeAsset(
+                    name=asset_properties["name"],
+                    icon=asset_properties["icon"],
+                    file_name=asset_properties["fileName"],
+                )
+            )
         recipe.assets = assets
         return recipe
 
@@ -413,9 +423,7 @@ def kptncook_to_mealie_steps(steps, api_key):
     return mealie_instructions
 
 
-def kptncook_to_mealie(
-    kcin: KptnCookRecipe, api_key: str = settings.kptncook_api_key
-) -> RecipeWithImage:
+def kptncook_to_mealie(kcin: KptnCookRecipe, api_key: str) -> RecipeWithImage:
     kwargs = {
         "name": kcin.localized_title.de,
         "notes": [

kptncook/paprika.py

@@ -25,7 +25,6 @@
 from jinja2 import Template
 from unidecode import unidecode
 
-from kptncook.config import settings
 from kptncook.models import Image, Recipe
 
 PAPRIKA_RECIPE_TEMPLATE = """{
@@ -70,6 +69,9 @@ class PaprikaExporter:
     template = Template(PAPRIKA_RECIPE_TEMPLATE, trim_blocks=True)
     unescaped_newline = re.compile(r"(?<!\\)\n")
 
+    def __init__(self, kptncook_api_key):
+        self.kptncook_api_key = kptncook_api_key
+
     def export(self, recipes: list[Recipe]) -> str:
         export_data = self.get_export_data(recipes=recipes)
         filename = self.get_export_filename(export_data=export_data, recipes=recipes)
@@ -155,7 +157,7 @@ def get_cover_img_as_base64_string(
         cover = self.get_cover(image_list=recipe.image_list)
         if cover is None:
             raise ValueError("No cover image found")
-        cover_url = recipe.get_image_url(api_key=settings.kptncook_api_key)
+        cover_url = recipe.get_image_url(api_key=self.kptncook_api_key)
         if not isinstance(cover_url, str):
             raise ValueError("Cover URL must be a string")
         try:

notebooks/api_search_tests.ipynb

@@ -17,7 +17,11 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "client = KptnCookClient()"
+    "from kptncook import Settings\n",
+    "\n",
+    "settings = Settings()\n",
+    "client = KptnCookClient(base_url=settings.kptncook_api_url, api_key=settings.kptncook_api_key,\n",
+    "                          access_token=settings.kptncook_access_token)"
    ]
   },
   {

notebooks/from_kptncook_to_mealie.ipynb

@@ -11,11 +11,13 @@
     "\n",
     "from pathlib import Path\n",
     "\n",
-    "from kptncook.config import settings\n",
+    "from kptncook.config import Settings\n",
     "from kptncook.models import Recipe\n",
     "from kptncook.mealie import kptncook_to_mealie\n",
     "\n",
-    "from kptncook.mealie import MealieApiClient"
+    "from kptncook.mealie import MealieApiClient\n",
+    "\n",
+    "settings = Settings()"
    ]
   },
   {

notebooks/get_or_create_units_from_ingredients.ipynb

@@ -147,9 +147,7 @@
    "id": "70775596-9500-490e-a4fd-8793a9332388",
    "metadata": {},
    "outputs": [],
-   "source": [
-    "mealie_recipe = kptncook_to_mealie(recipe)"
-   ]
+   "source": "mealie_recipe = kptncook_to_mealie(recipe, settings.kptncook_api_key)"
   },
   {
    "cell_type": "code",

notebooks/issue_25_broken_urls.ipynb

@@ -40,7 +40,12 @@
     }
    ],
    "source": [
-    "client = KptnCookClient()\n",
+    "from kptncook import Settings\n",
+    "\n",
+    "settings = Settings()\n",
+    "client = KptnCookClient(base_url=settings.kptncook_api_url, api_key=settings.kptncook_api_key,\n",
+    "                        access_token=settings.kptncook_access_token)\n",
+    "\n",
     "all_recipes = client.list_today()\n",
     "len(all_recipes)"
    ]

notebooks/test_new_headers.ipynb

@@ -23,7 +23,11 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "client = KptnCookClient()"
+    "from kptncook import Settings\n",
+    "\n",
+    "settings = Settings()\n",
+    "client = KptnCookClient(base_url=settings.kptncook_api_url, api_key=settings.kptncook_api_key,\n",
+    "                        access_token=settings.kptncook_access_token)"
    ]
   },
   {

tests/api_test.py

@@ -3,10 +3,14 @@
 from kptncook.api import KptnCookClient
 
 
-def test_client_to_url():
+def test_client_to_url(test_settings):
     base_url = Url(
         "https://mobile.kptncook.com"
     )  # make sure urljoin works with pydantic URLs
-    client = KptnCookClient(base_url=base_url)
+    client = KptnCookClient(
+        base_url=base_url,
+        api_key=test_settings.kptncook_api_key,
+        access_token=test_settings.kptncook_access_token,
+    )
     assert client.to_url("/recipes") == "https://mobile.kptncook.com/recipes"
     assert client.to_url("recipes") == "https://mobile.kptncook.com/recipes"

tests/config_test.py

@@ -0,0 +1,21 @@
+import pydantic
+import pytest
+
+from tests.conftest import MockSettings
+
+
+def test_config_accepts_mealie_username_password():
+    MockSettings(
+        mealie_username="test", mealie_password="password", kptncook_api_key="test"
+    )
+
+
+def test_config_accepts_mealie_token():
+    MockSettings(mealie_api_token="test", kptncook_api_key="test")
+
+
+def test_config_rejects_empty_mealie_auth():
+    with pytest.raises(pydantic.ValidationError) as exception_info:
+        MockSettings(kptncook_api_key="test")
+
+    assert "must specify either" in str(exception_info.value)