mirror of
https://github.com/mealie-recipes/mealie.git
synced 2025-12-27 20:55:12 -05:00
refactor: ♻️ rewrite migrations frontend/backend (#841)
* refactor(frontend): ♻️ rewrite migrations UI * refactor(backend): ♻️ rewrite recipe migrations * remove vue-demi Co-authored-by: hay-kot <hay-kot@pm.me>
This commit is contained in:
Hayden
GitHub
@@ -1,122 +1,134 @@
|
||||
import json
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
from typing import Any, Callable, Optional
|
||||
|
||||
import yaml
|
||||
from pydantic import BaseModel
|
||||
from typing import Tuple
|
||||
|
||||
from mealie.core import root_logger
|
||||
from mealie.db.database import get_database
|
||||
from mealie.schema.admin import MigrationImport
|
||||
from mealie.db.database import Database
|
||||
from mealie.schema.recipe import Recipe
|
||||
from mealie.schema.user.user import PrivateUser
|
||||
from mealie.services.image import image
|
||||
from mealie.schema.reports.reports import (
|
||||
ReportCategory,
|
||||
ReportCreate,
|
||||
ReportEntryCreate,
|
||||
ReportOut,
|
||||
ReportSummary,
|
||||
ReportSummaryStatus,
|
||||
)
|
||||
from mealie.services.scraper import cleaner
|
||||
from mealie.utils.unzip import unpack_zip
|
||||
|
||||
logger = root_logger.get_logger()
|
||||
from .._base_service import BaseService
|
||||
from .utils.migration_alias import MigrationAlias
|
||||
|
||||
|
||||
class MigrationAlias(BaseModel):
|
||||
"""A datatype used by MigrationBase to pre-process a recipe dictionary to rewrite
|
||||
the alias key in the dictionary, if it exists, to the key. If set a `func` attribute
|
||||
will be called on the value before assigning the value to the new key
|
||||
"""
|
||||
class BaseMigrator(BaseService):
|
||||
key_aliases: list[MigrationAlias]
|
||||
|
||||
key: str
|
||||
alias: str
|
||||
func: Optional[Callable] = None
|
||||
report_entries: list[ReportEntryCreate]
|
||||
report_id: int
|
||||
report: ReportOut
|
||||
|
||||
def __init__(self, archive: Path, db: Database, session, user_id: int, group_id: int):
|
||||
self.archive = archive
|
||||
self.db = db
|
||||
self.session = session
|
||||
self.user_id = user_id
|
||||
self.group_id = group_id
|
||||
|
||||
class MigrationBase(BaseModel):
|
||||
migration_report: list[MigrationImport] = []
|
||||
migration_file: Path
|
||||
session: Optional[Any]
|
||||
key_aliases: Optional[list[MigrationAlias]]
|
||||
self.report_entries = []
|
||||
|
||||
user: PrivateUser
|
||||
self.logger = root_logger.get_logger()
|
||||
|
||||
@property
|
||||
def db(self):
|
||||
return get_database(self.session)
|
||||
super().__init__()
|
||||
|
||||
@property
|
||||
def temp_dir(self) -> TemporaryDirectory:
|
||||
"""unpacks the migration_file into a temporary directory
|
||||
that can be used as a context manager.
|
||||
def _migrate(self) -> None:
|
||||
raise NotImplementedError
|
||||
|
||||
Returns:
|
||||
TemporaryDirectory:
|
||||
def _create_report(self, report_name: str) -> None:
|
||||
report_to_save = ReportCreate(
|
||||
name=report_name,
|
||||
category=ReportCategory.migration,
|
||||
status=ReportSummaryStatus.in_progress,
|
||||
group_id=self.group_id,
|
||||
)
|
||||
|
||||
self.report = self.db.group_reports.create(report_to_save)
|
||||
self.report_id = self.report.id
|
||||
|
||||
def _save_all_entries(self) -> None:
|
||||
|
||||
is_success = True
|
||||
is_failure = True
|
||||
|
||||
for entry in self.report_entries:
|
||||
if is_failure and entry.success:
|
||||
is_failure = False
|
||||
|
||||
if is_success and not entry.success:
|
||||
is_success = False
|
||||
|
||||
self.db.group_report_entries.create(entry)
|
||||
|
||||
if is_success:
|
||||
self.report.status = ReportSummaryStatus.success
|
||||
|
||||
if is_failure:
|
||||
self.report.status = ReportSummaryStatus.failure
|
||||
|
||||
if not is_success and not is_failure:
|
||||
self.report.status = ReportSummaryStatus.partial
|
||||
|
||||
self.db.group_reports.update(self.report.id, self.report)
|
||||
|
||||
def migrate(self, report_name: str) -> ReportSummary:
|
||||
self._create_report(report_name)
|
||||
self._migrate()
|
||||
self._save_all_entries()
|
||||
return self.db.group_reports.get(self.report_id)
|
||||
|
||||
def import_recipes_to_database(self, validated_recipes: list[Recipe]) -> list[Tuple[str, bool]]:
|
||||
"""
|
||||
return unpack_zip(self.migration_file)
|
||||
|
||||
@staticmethod
|
||||
def json_reader(json_file: Path) -> dict:
|
||||
with open(json_file, "r") as f:
|
||||
return json.loads(f.read())
|
||||
|
||||
@staticmethod
|
||||
def yaml_reader(yaml_file: Path) -> dict:
|
||||
"""A helper function to read in a yaml file from a Path. This assumes that the
|
||||
first yaml document is the recipe data and the second, if exists, is the description.
|
||||
Used as a single access point to process a list of Recipe objects into the
|
||||
database in a predictable way. If an error occurs the session is rolled back
|
||||
and the process will continue. All import information is appended to the
|
||||
'migration_report' attribute to be returned to the frontend for display.
|
||||
|
||||
Args:
|
||||
yaml_file (Path): Path to yaml file
|
||||
|
||||
Returns:
|
||||
dict: representing the yaml file as a dictionary
|
||||
validated_recipes (list[Recipe]):
|
||||
"""
|
||||
with open(yaml_file, "r") as f:
|
||||
contents = f.read().split("---")
|
||||
recipe_data = {}
|
||||
for _, document in enumerate(contents):
|
||||
|
||||
# Check if None or Empty String
|
||||
if document is None or document == "":
|
||||
continue
|
||||
return_vars = []
|
||||
|
||||
# Check if 'title:' present
|
||||
elif "title:" in document:
|
||||
recipe_data.update(yaml.safe_load(document))
|
||||
for recipe in validated_recipes:
|
||||
|
||||
else:
|
||||
recipe_data["description"] = document
|
||||
recipe.user_id = self.user_id
|
||||
recipe.group_id = self.group_id
|
||||
|
||||
return recipe_data
|
||||
exception = ""
|
||||
status = False
|
||||
try:
|
||||
self.db.recipes.create(recipe)
|
||||
status = True
|
||||
|
||||
@staticmethod
|
||||
def glob_walker(directory: Path, glob_str: str, return_parent=True) -> list[Path]: # TODO:
|
||||
"""A Helper function that will return the glob matches for the temporary directotry
|
||||
that was unpacked and passed in as the `directory` parameter. If `return_parent` is
|
||||
True the return Paths will be the parent directory for the file that was matched. If
|
||||
false the file itself will be returned.
|
||||
except Exception as inst:
|
||||
exception = inst
|
||||
self.logger.exception(inst)
|
||||
self.session.rollback()
|
||||
|
||||
Args:
|
||||
directory (Path): Path to search directory
|
||||
glob_str ([type]): glob style match string
|
||||
return_parent (bool, optional): To return parent directory of match. Defaults to True.
|
||||
|
||||
Returns:
|
||||
list[Path]:
|
||||
"""
|
||||
directory = directory if isinstance(directory, Path) else Path(directory)
|
||||
matches = []
|
||||
for match in directory.glob(glob_str):
|
||||
if return_parent:
|
||||
matches.append(match.parent)
|
||||
if status:
|
||||
message = f"Imported {recipe.name} successfully"
|
||||
else:
|
||||
matches.append(match)
|
||||
message = f"Failed to import {recipe.name}"
|
||||
|
||||
return matches
|
||||
return_vars.append((recipe.slug, status))
|
||||
|
||||
@staticmethod
|
||||
def import_image(src: Path, dest_slug: str):
|
||||
"""Read the successful migrations attribute and for each import the image
|
||||
appropriately into the image directory. Minification is done in mass
|
||||
after the migration occurs.
|
||||
"""
|
||||
image.write_image(dest_slug, src, extension=src.suffix)
|
||||
self.report_entries.append(
|
||||
ReportEntryCreate(
|
||||
report_id=self.report_id,
|
||||
success=status,
|
||||
message=message,
|
||||
exception=str(exception),
|
||||
)
|
||||
)
|
||||
|
||||
return return_vars
|
||||
|
||||
def rewrite_alias(self, recipe_dict: dict) -> dict:
|
||||
"""A helper function to reassign attributes by an alias using a list
|
||||
@@ -137,7 +149,6 @@ class MigrationBase(BaseModel):
|
||||
try:
|
||||
prop_value = recipe_dict.pop(alias.alias)
|
||||
except KeyError:
|
||||
logger.info(f"Key {alias.alias} Not Found. Skipping...")
|
||||
continue
|
||||
|
||||
if alias.func:
|
||||
@@ -147,7 +158,7 @@ class MigrationBase(BaseModel):
|
||||
|
||||
return recipe_dict
|
||||
|
||||
def clean_recipe_dictionary(self, recipe_dict) -> Recipe:
|
||||
def clean_recipe_dictionary(self, recipe_dict: dict) -> Recipe:
|
||||
"""
|
||||
Calls the rewrite_alias function and the Cleaner.clean function on a
|
||||
dictionary and returns the result unpacked into a Recipe object
|
||||
@@ -156,33 +167,3 @@ class MigrationBase(BaseModel):
|
||||
recipe_dict = cleaner.clean(recipe_dict, url=recipe_dict.get("org_url", None))
|
||||
|
||||
return Recipe(**recipe_dict)
|
||||
|
||||
def import_recipes_to_database(self, validated_recipes: list[Recipe]) -> None:
|
||||
"""
|
||||
Used as a single access point to process a list of Recipe objects into the
|
||||
database in a predictable way. If an error occurs the session is rolled back
|
||||
and the process will continue. All import information is appended to the
|
||||
'migration_report' attribute to be returned to the frontend for display.
|
||||
|
||||
Args:
|
||||
validated_recipes (list[Recipe]):
|
||||
"""
|
||||
|
||||
for recipe in validated_recipes:
|
||||
|
||||
recipe.user_id = self.user.id
|
||||
recipe.group_id = self.user.group_id
|
||||
|
||||
exception = ""
|
||||
status = False
|
||||
try:
|
||||
self.db.recipes.create(recipe.dict())
|
||||
status = True
|
||||
|
||||
except Exception as inst:
|
||||
exception = inst
|
||||
logger.exception(inst)
|
||||
self.session.rollback()
|
||||
|
||||
import_status = MigrationImport(slug=recipe.slug, name=recipe.name, status=status, exception=str(exception))
|
||||
self.migration_report.append(import_status)
|
||||
|
||||
Reference in New Issue
Block a user