github/mealie
1
0
Fork 0
mirror of https://github.com/hay-kot/mealie.git synced 2025-08-22 14:33:33 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions 20

refactor settings to class for testing

Browse source
This commit is contained in:
hay-kot 2021-03-30 09:43:00 -08:00
commit deeb8bd707
33 changed files with 292 additions and 315 deletions
Download patch file Download diff file Expand all files Collapse all files

8
makefile
View file

@ -73,4 +73,10 @@ docker-prod: ## Build and Start Docker Production Stack
code-gen: ## Run Code-Gen Scripts
poetry run python dev/scripts/app_routes_gen.py
poetry run python dev/scripts/app_routes_gen.py
coverage: ## check code coverage quickly with the default Python
poetry run pytest
coverage report -m
coverage html
$(BROWSER) htmlcov/index.html

10
mealie/app.py
View file

@ -3,20 +3,20 @@ from fastapi import FastAPI
from fastapi.logger import logger
# import utils.startup as startup
from mealie.core.config import APP_VERSION, PORT, docs_url, redoc_url
from mealie.core.config import APP_VERSION, settings
from mealie.routes import backup_routes, debug_routes, migration_routes, theme_routes
from mealie.routes.site_settings import all_settings
from mealie.routes.groups import groups
from mealie.routes.mealplans import mealplans
from mealie.routes.recipe import all_recipe_routes, category_routes, recipe_crud_routes, tag_routes
from mealie.routes.site_settings import all_settings
from mealie.routes.users import users
app = FastAPI(
title="Mealie",
description="A place for all your recipes",
version=APP_VERSION,
docs_url=docs_url,
redoc_url=redoc_url,
docs_url=settings.DOCS_URL,
redoc_url=settings.REDOC_URL,
)
@ -55,7 +55,7 @@ def main():
uvicorn.run(
"app:app",
host="0.0.0.0",
port=PORT,
port=settings.API_PORT,
reload=True,
reload_dirs=["mealie"],
debug=True,

156
mealie/core/config.py
View file

@ -8,80 +8,23 @@ APP_VERSION = "v0.4.0"
DB_VERSION = "v0.4.0"
CWD = Path(__file__).parent
BASE_DIR = CWD.parent.parent
def ensure_dirs():
for dir in REQUIRED_DIRS:
dir.mkdir(parents=True, exist_ok=True)
# Register ENV
ENV = CWD.joinpath(".env") # ! I'm Broken Fix Me!
ENV = BASE_DIR.joinpath(".env")
dotenv.load_dotenv(ENV)
PRODUCTION = os.environ.get("ENV")
# General
PORT = int(os.getenv("mealie_port", 9000))
API = os.getenv("api_docs", True)
def determine_data_dir(production: bool) -> Path:
global CWD
if production:
return Path("/app/data")
if API:
docs_url = "/docs"
redoc_url = "/redoc"
else:
docs_url = None
redoc_url = None
# Helpful Globals
DATA_DIR = CWD.parent.parent.joinpath("dev", "data")
if PRODUCTION:
DATA_DIR = Path("/app/data")
WEB_PATH = CWD.joinpath("dist")
IMG_DIR = DATA_DIR.joinpath("img")
BACKUP_DIR = DATA_DIR.joinpath("backups")
DEBUG_DIR = DATA_DIR.joinpath("debug")
MIGRATION_DIR = DATA_DIR.joinpath("migration")
NEXTCLOUD_DIR = MIGRATION_DIR.joinpath("nextcloud")
CHOWDOWN_DIR = MIGRATION_DIR.joinpath("chowdown")
TEMPLATE_DIR = DATA_DIR.joinpath("templates")
USER_DIR = DATA_DIR.joinpath("users")
SQLITE_DIR = DATA_DIR.joinpath("db")
RECIPE_DATA_DIR = DATA_DIR.joinpath("recipes")
TEMP_DIR = DATA_DIR.joinpath(".temp")
REQUIRED_DIRS = [
DATA_DIR,
IMG_DIR,
BACKUP_DIR,
DEBUG_DIR,
MIGRATION_DIR,
TEMPLATE_DIR,
SQLITE_DIR,
NEXTCLOUD_DIR,
CHOWDOWN_DIR,
RECIPE_DATA_DIR,
USER_DIR,
]
ensure_dirs()
LOGGER_FILE = DATA_DIR.joinpath("mealie.log")
return CWD.parent.parent.joinpath("dev", "data")
# DATABASE ENV
SQLITE_FILE = None
DATABASE_TYPE = os.getenv("DB_TYPE", "sqlite")
if DATABASE_TYPE == "sqlite":
USE_SQL = True
SQLITE_FILE = SQLITE_DIR.joinpath(f"mealie_{DB_VERSION}.sqlite")
else:
raise Exception("Unable to determine database type. Acceptible options are 'sqlite' ")
def determine_secrets() -> str:
if not PRODUCTION:
def determine_secrets(production: bool) -> str:
if not production:
return "shh-secret-test-key"
secrets_file = DATA_DIR.joinpath(".secret")
@ -90,22 +33,75 @@ def determine_secrets() -> str:
return f.read()
else:
with open(secrets_file, "w") as f:
f.write(secrets.token_hex(32))
new_secret = secrets.token_hex(32)
f.write(new_secret)
return new_secret
SECRET = "determine_secrets()"
class AppDirectories:
def __init__(self, cwd, data_dir) -> None:
self.WEB_PATH = cwd.joinpath("dist")
self.IMG_DIR = data_dir.joinpath("img")
self.BACKUP_DIR = data_dir.joinpath("backups")
self.DEBUG_DIR = data_dir.joinpath("debug")
self.MIGRATION_DIR = data_dir.joinpath("migration")
self.NEXTCLOUD_DIR = self.MIGRATION_DIR.joinpath("nextcloud")
self.CHOWDOWN_DIR = self.MIGRATION_DIR.joinpath("chowdown")
self.TEMPLATE_DIR = data_dir.joinpath("templates")
self.USER_DIR = data_dir.joinpath("users")
self.SQLITE_DIR = data_dir.joinpath("db")
self.RECIPE_DATA_DIR = data_dir.joinpath("recipes")
self.TEMP_DIR = data_dir.joinpath(".temp")
# Mongo Database
DEFAULT_GROUP = os.getenv("DEFAULT_GROUP", "Home")
DEFAULT_PASSWORD = os.getenv("DEFAULT_PASSWORD", "MyPassword")
self.ensure_directories()
# Database
MEALIE_DB_NAME = os.getenv("mealie_db_name", "mealie")
DB_USERNAME = os.getenv("db_username", "root")
DB_PASSWORD = os.getenv("db_password", "example")
DB_HOST = os.getenv("db_host", "mongo")
DB_PORT = os.getenv("db_port", 27017)
def ensure_directories(self):
required_dirs = [
self.IMG_DIR,
self.BACKUP_DIR,
self.DEBUG_DIR,
self.MIGRATION_DIR,
self.TEMPLATE_DIR,
self.SQLITE_DIR,
self.NEXTCLOUD_DIR,
self.CHOWDOWN_DIR,
self.RECIPE_DATA_DIR,
self.USER_DIR,
]
# SFTP Email Stuff - For use Later down the line!
SFTP_USERNAME = os.getenv("sftp_username", None)
SFTP_PASSWORD = os.getenv("sftp_password", None)
for dir in required_dirs:
dir.mkdir(parents=True, exist_ok=True)
class AppSettings:
def __init__(self, app_dirs: AppDirectories) -> None:
global DB_VERSION
self.PRODUCTION = bool(os.environ.get("ENV"))
self.API_PORT = int(os.getenv("API_PORT", 9000))
self.API = bool(os.getenv("API_DOCS", True))
self.DOCS_URL = "/docs" if self.API else None
self.REDOC_URL = "/redoc" if self.API else None
self.SECRET = determine_secrets(self.PRODUCTION)
self.DATABASE_TYPE = os.getenv("DB_TYPE", "sqlite")
# Used to Set SQLite File Version
self.SQLITE_FILE = None
if self.DATABASE_TYPE == "sqlite":
self.SQLITE_FILE = app_dirs.SQLITE_DIR.joinpath(f"mealie_{DB_VERSION}.sqlite")
else:
raise Exception("Unable to determine database type. Acceptible options are 'sqlite' ")
self.DEFAULT_GROUP = os.getenv("DEFAULT_GROUP", "Home")
self.DEFAULT_PASSWORD = os.getenv("DEFAULT_PASSWORD", "MyPassword")
# Not Used!
self.SFTP_USERNAME = os.getenv("SFTP_USERNAME", None)
self.SFTP_PASSWORD = os.getenv("SFTP_PASSWORD", None)
# General
DATA_DIR = determine_data_dir(PRODUCTION)
LOGGER_FILE = DATA_DIR.joinpath("mealie.log")
app_dirs = AppDirectories(CWD, DATA_DIR)
settings = AppSettings(app_dirs)

4
mealie/core/security.py
View file

@ -2,7 +2,7 @@ from datetime import datetime, timedelta
from mealie.schema.user import UserInDB
from jose import jwt
from mealie.core.config import SECRET
from mealie.core.config import settings
from mealie.db.database import db
from passlib.context import CryptContext
@ -17,7 +17,7 @@ def create_access_token(data: dict(), expires_delta: timedelta = None) -> str:
else:
expire = datetime.utcnow() + timedelta(minutes=120)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET, algorithm=ALGORITHM)
return jwt.encode(to_encode, settings.SECRET, algorithm=ALGORITHM)
def authenticate_user(session, email: str, password: str) -> UserInDB:

25
mealie/db/db_base.py
View file

@ -1,11 +1,10 @@
from typing import List
from mealie.db.models.model_base import SqlAlchemyBase
from pydantic import BaseModel
from sqlalchemy.orm import load_only
from sqlalchemy.orm.session import Session
from mealie.db.models.model_base import SqlAlchemyBase
class BaseDocument:
def __init__(self) -> None:
@ -21,12 +20,12 @@ class BaseDocument:
if self.orm_mode:
return [self.schema.from_orm(x) for x in session.query(self.sql_model).limit(limit).all()]
list = [x.dict() for x in session.query(self.sql_model).limit(limit).all()]
# list = [x.dict() for x in session.query(self.sql_model).limit(limit).all()]
if limit == 1:
return list[0]
# if limit == 1:
# return list[0]
return list
# return list
def get_all_limit_columns(self, session: Session, fields: List[str], limit: int = None) -> List[SqlAlchemyBase]:
"""Queries the database for the selected model. Restricts return responses to the
@ -95,6 +94,7 @@ class BaseDocument:
return self.schema.from_orm(result[0])
except IndexError:
return None
return [self.schema.from_orm(x) for x in result]
def create(self, session: Session, document: dict) -> BaseModel:
@ -111,10 +111,8 @@ class BaseDocument:
session.add(new_document)
session.commit()
if self.orm_mode:
return self.schema.from_orm(new_document)
return self.schema.from_orm(new_document)
return new_document.dict()
def update(self, session: Session, match_value: str, new_data: str) -> BaseModel:
"""Update a database entry.
@ -131,13 +129,10 @@ class BaseDocument:
entry = self._query_one(session=session, match_value=match_value)
entry.update(session=session, **new_data)
if self.orm_mode:
session.commit()
return self.schema.from_orm(entry)
return_data = entry.dict()
session.commit()
return return_data
return self.schema.from_orm(entry)
def delete(self, session: Session, primary_key_value) -> dict:
result = session.query(self.sql_model).filter_by(**{self.primary_key: primary_key_value}).one()

9
mealie/db/db_setup.py
View file

@ -1,15 +1,12 @@
from mealie.core.config import SQLITE_FILE, USE_SQL
from mealie.core.config import settings
from sqlalchemy.orm.session import Session
from mealie.db.models.db_session import sql_global_init
sql_exists = True
if USE_SQL:
sql_exists = SQLITE_FILE.is_file()
SessionLocal = sql_global_init(SQLITE_FILE)
else:
raise Exception("Cannot identify database type")
sql_exists = settings.SQLITE_FILE.is_file()
SessionLocal = sql_global_init(settings.SQLITE_FILE)
def create_session() -> Session:

8
mealie/db/init_db.py
View file

@ -1,5 +1,5 @@
from fastapi.logger import logger
from mealie.core.config import DEFAULT_GROUP, DEFAULT_PASSWORD
from mealie.core.config import settings
from mealie.core.security import get_password_hash
from mealie.db.database import db
from mealie.db.db_setup import create_session, sql_exists
@ -30,7 +30,7 @@ def default_settings_init(session: Session):
def default_group_init(session: Session):
default_group = {"name": DEFAULT_GROUP}
default_group = {"name": settings.DEFAULT_GROUP}
logger.info("Generating Default Group")
db.groups.create(session, default_group)
@ -39,8 +39,8 @@ def default_user_init(session: Session):
default_user = {
"full_name": "Change Me",
"email": "changeme@email.com",
"password": get_password_hash(DEFAULT_PASSWORD),
"group": DEFAULT_GROUP,
"password": get_password_hash(settings.DEFAULT_PASSWORD),
"group": settings.DEFAULT_GROUP,
"admin": True,
}

4
mealie/db/models/group.py
View file

@ -1,7 +1,7 @@
import sqlalchemy as sa
import sqlalchemy.orm as orm
from fastapi.logger import logger
from mealie.core.config import DEFAULT_GROUP
from mealie.core.config import settings
from mealie.db.models.model_base import BaseMixins, SqlAlchemyBase
from mealie.db.models.recipe.category import Category, group2categories
from sqlalchemy.orm.session import Session
@ -63,7 +63,7 @@ class Group(SqlAlchemyBase, BaseMixins):
return item
@staticmethod
def create_if_not_exist(session, name: str = DEFAULT_GROUP):
def create_if_not_exist(session, name: str = settings.DEFAULT_GROUP):
result = session.query(Group).filter(Group.name == name).one_or_none()
if result:
logger.info("Group exists, associating recipe")

43
mealie/db/models/model_base.py
View file

@ -1,47 +1,8 @@
from typing import List
import sqlalchemy.ext.declarative as dec
from sqlalchemy.orm.session import Session
SqlAlchemyBase = dec.declarative_base()
class BaseMixins:
@staticmethod
def _sql_remove_list(session: Session, list_of_tables: list, parent_id):
for table in list_of_tables:
session.query(table).filter(parent_id == parent_id).delete()
@staticmethod
def _flatten_dict(list_of_dict: List[dict]):
finalMap = {}
for d in list_of_dict:
finalMap.update(d.dict())
return finalMap
# ! Don't use!
def update_generics(func):
"""An experimental function that does the initial work of updating attributes on a class
and passing "complex" data types recuresively to an "self.update()" function if one exists.
Args:
func ([type]): [description]
"""
def wrapper(class_object, session, new_data: dict):
complex_attributed = {}
for key, value in new_data.items():
attribute = getattr(class_object, key, None)
if attribute and isinstance(attribute, SqlAlchemyBase):
attribute.update(session, value)
elif attribute:
setattr(class_object, key, value)
func(class_object, complex_attributed)
return wrapper
def _pass_on_me():
pass

6
mealie/db/models/users.py
View file

@ -1,4 +1,4 @@
from mealie.core.config import DEFAULT_GROUP
from mealie.core.config import settings
from mealie.db.models.group import Group
from mealie.db.models.model_base import BaseMixins, SqlAlchemyBase
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, orm
@ -26,12 +26,12 @@ class User(SqlAlchemyBase, BaseMixins):
full_name,
email,
password,
group: str = DEFAULT_GROUP,
group: str = settings.DEFAULT_GROUP,
admin=False,
id=None,
) -> None:
group = group if group else DEFAULT_GROUP
group = group or settings.DEFAULT_GROUP
self.full_name = full_name
self.email = email
self.group = Group.get_ref(session, group)

12
mealie/routes/backup_routes.py
View file

@ -2,7 +2,7 @@ import operator
import shutil
from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
from mealie.core.config import BACKUP_DIR, TEMPLATE_DIR
from mealie.core.config import app_dirs
from mealie.db.db_setup import generate_session
from mealie.routes.deps import get_current_user
from mealie.schema.backup import BackupJob, ImportJob, Imports, LocalBackup
@ -19,11 +19,11 @@ router = APIRouter(prefix="/api/backups", tags=["Backups"], dependencies=[Depend
def available_imports():
"""Returns a list of avaiable .zip files for import into Mealie."""
imports = []
for archive in BACKUP_DIR.glob("*.zip"):
for archive in app_dirs.app_dirs.BACKUP_DIR.glob("*.zip"):
backup = LocalBackup(name=archive.name, date=archive.stat().st_ctime)
imports.append(backup)
templates = [template.name for template in TEMPLATE_DIR.glob("*.*")]
templates = [template.name for template in app_dirs.TEMPLATE_DIR.glob("*.*")]
imports.sort(key=operator.attrgetter("date"), reverse=True)
return Imports(imports=imports, templates=templates)
@ -55,7 +55,7 @@ def export_database(data: BackupJob, session: Session = Depends(generate_session
@router.post("/upload")
def upload_backup_file(archive: UploadFile = File(...)):
""" Upload a .zip File to later be imported into Mealie """
dest = BACKUP_DIR.joinpath(archive.filename)
dest = app_dirs.BACKUP_DIR.joinpath(archive.filename)
with dest.open("wb") as buffer:
shutil.copyfileobj(archive.file, buffer)
@ -69,7 +69,7 @@ def upload_backup_file(archive: UploadFile = File(...)):
@router.get("/{file_name}/download")
async def download_backup_file(file_name: str):
""" Upload a .zip File to later be imported into Mealie """
file = BACKUP_DIR.joinpath(file_name)
file = app_dirs.BACKUP_DIR.joinpath(file_name)
if file.is_file:
return FileResponse(file, media_type="application/octet-stream", filename=file_name)
@ -100,7 +100,7 @@ def delete_backup(file_name: str):
""" Removes a database backup from the file system """
try:
BACKUP_DIR.joinpath(file_name).unlink()
app_dirs.BACKUP_DIR.joinpath(file_name).unlink()
except:
HTTPException(
status_code=400,

4
mealie/routes/debug_routes.py
View file

@ -1,7 +1,7 @@
import json
from fastapi import APIRouter, Depends
from mealie.core.config import APP_VERSION, DEBUG_DIR, LOGGER_FILE
from mealie.core.config import APP_VERSION, LOGGER_FILE, app_dirs
from mealie.routes.deps import get_current_user
router = APIRouter(prefix="/api/debug", tags=["Debug"], dependencies=[Depends(get_current_user)])
@ -17,7 +17,7 @@ async def get_mealie_version():
async def get_last_recipe_json():
""" Doc Str """
with open(DEBUG_DIR.joinpath("last_recipe.json"), "r") as f:
with open(app_dirs.DEBUG_DIR.joinpath("last_recipe.json"), "r") as f:
return json.loads(f.read())

4
mealie/routes/deps.py
View file

@ -1,7 +1,7 @@
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from mealie.core.config import SECRET
from mealie.core.config import settings
from mealie.db.database import db
from mealie.db.db_setup import generate_session
from mealie.schema.auth import TokenData
@ -18,7 +18,7 @@ async def get_current_user(token: str = Depends(oauth2_scheme), session=Depends(
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET, algorithms=[ALGORITHM])
payload = jwt.decode(token, settings.SECRET, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception

12
mealie/routes/migration_routes.py
View file

@ -3,7 +3,7 @@ import shutil
from typing import List
from fastapi import APIRouter, Depends, File, UploadFile
from mealie.core.config import MIGRATION_DIR
from mealie.core.config import app_dirs
from mealie.db.db_setup import generate_session
from mealie.routes.deps import get_current_user
from mealie.schema.migration import MigrationFile, Migrations
@ -20,8 +20,8 @@ def get_avaiable_nextcloud_imports():
""" Returns a list of avaiable directories that can be imported into Mealie """
response_data = []
migration_dirs = [
MIGRATION_DIR.joinpath("nextcloud"),
MIGRATION_DIR.joinpath("chowdown"),
app_dirs.MIGRATION_DIR.joinpath("nextcloud"),
app_dirs.MIGRATION_DIR.joinpath("chowdown"),
]
for directory in migration_dirs:
migration = Migrations(type=directory.stem)
@ -39,7 +39,7 @@ def get_avaiable_nextcloud_imports():
@router.post("/{type}/{file_name}/import")
def import_nextcloud_directory(type: str, file_name: str, session: Session = Depends(generate_session)):
""" Imports all the recipes in a given directory """
file_path = MIGRATION_DIR.joinpath(type, file_name)
file_path = app_dirs.MIGRATION_DIR.joinpath(type, file_name)
if type == "nextcloud":
return nextcloud_migrate(session, file_path)
elif type == "chowdown":
@ -52,7 +52,7 @@ def import_nextcloud_directory(type: str, file_name: str, session: Session = Dep
def delete_migration_data(type: str, file_name: str):
""" Removes migration data from the file system """
remove_path = MIGRATION_DIR.joinpath(type, file_name)
remove_path = app_dirs.MIGRATION_DIR.joinpath(type, file_name)
if remove_path.is_file():
remove_path.unlink()
@ -67,7 +67,7 @@ def delete_migration_data(type: str, file_name: str):
@router.post("/{type}/upload")
def upload_nextcloud_zipfile(type: str, archive: UploadFile = File(...)):
""" Upload a .zip File to later be imported into Mealie """
dir = MIGRATION_DIR.joinpath(type)
dir = app_dirs.MIGRATION_DIR.joinpath(type)
dir.mkdir(parents=True, exist_ok=True)
dest = dir.joinpath(archive.filename)

12
mealie/routes/users/crud.py
View file

@ -4,7 +4,7 @@ from datetime import timedelta
from fastapi import APIRouter, Depends, File, UploadFile
from fastapi.responses import FileResponse
from mealie.core import security
from mealie.core.config import DEFAULT_PASSWORD, USER_DIR
from mealie.core.config import settings, app_dirs
from mealie.core.security import get_password_hash, verify_password
from mealie.db.database import db
from mealie.db.db_setup import generate_session
@ -65,7 +65,7 @@ async def reset_user_password(
session: Session = Depends(generate_session),
):
new_password = get_password_hash(DEFAULT_PASSWORD)
new_password = get_password_hash(settings.DEFAULT_PASSWORD)
db.users.update_password(session, id, new_password)
return SnackResponse.success("Users Password Reset")
@ -92,7 +92,7 @@ async def update_user(
@router.get("/{id}/image")
async def get_user_image(id: str):
""" Returns a users profile picture """
user_dir = USER_DIR.joinpath(id)
user_dir = app_dirs.USER_DIR.joinpath(id)
for recipe_image in user_dir.glob("profile_image.*"):
return FileResponse(recipe_image)
else:
@ -109,14 +109,14 @@ async def update_user_image(
extension = profile_image.filename.split(".")[-1]
USER_DIR.joinpath(id).mkdir(parents=True, exist_ok=True)
app_dirs.USER_DIR.joinpath(id).mkdir(parents=True, exist_ok=True)
try:
[x.unlink() for x in USER_DIR.join(id).glob("profile_image.*")]
[x.unlink() for x in app_dirs.USER_DIR.join(id).glob("profile_image.*")]
except:
pass
dest = USER_DIR.joinpath(id, f"profile_image.{extension}")
dest = app_dirs.USER_DIR.joinpath(id, f"profile_image.{extension}")
with dest.open("wb") as buffer:
shutil.copyfileobj(profile_image.file, buffer)

4
mealie/schema/user.py
View file

@ -1,7 +1,7 @@
from typing import Optional
from fastapi_camelcase import CamelModel
from mealie.core.config import DEFAULT_GROUP
from mealie.core.config import settings
from mealie.db.models.group import Group
from mealie.db.models.users import User
from mealie.schema.category import CategoryBase
@ -40,7 +40,7 @@ class UserBase(CamelModel):
schema_extra = {
"fullName": "Change Me",
"email": "changeme@email.com",
"group": DEFAULT_GROUP,
"group": settings.DEFAULT_GROUP,
"admin": "false",
}

16
mealie/services/backups/exports.py
View file

@ -6,7 +6,7 @@ from typing import Union
from fastapi.logger import logger
from jinja2 import Template
from mealie.core.config import BACKUP_DIR, IMG_DIR, TEMP_DIR, TEMPLATE_DIR
from mealie.core.config import app_dirs
from mealie.db.database import db
from mealie.db.db_setup import create_session
from pydantic.main import BaseModel
@ -28,12 +28,12 @@ class ExportDatabase:
else:
export_tag = datetime.now().strftime("%Y-%b-%d")
self.main_dir = TEMP_DIR.joinpath(export_tag)
self.main_dir = app_dirs.TEMP_DIR.joinpath(export_tag)
self.img_dir = self.main_dir.joinpath("images")
self.templates_dir = self.main_dir.joinpath("templates")
try:
self.templates = [TEMPLATE_DIR.joinpath(x) for x in templates]
self.templates = [app_dirs.TEMPLATE_DIR.joinpath(x) for x in templates]
except:
self.templates = False
logger.info("No Jinja2 Templates Registered for Export")
@ -65,7 +65,7 @@ class ExportDatabase:
f.write(content)
def export_images(self):
for file in IMG_DIR.iterdir():
for file in app_dirs.IMG_DIR.iterdir():
shutil.copy(file, self.img_dir.joinpath(file.name))
def export_items(self, items: list[BaseModel], folder_name: str, export_list=True):
@ -87,10 +87,10 @@ class ExportDatabase:
f.write(json_data)
def finish_export(self):
zip_path = BACKUP_DIR.joinpath(f"{self.main_dir.name}")
zip_path = app_dirs.BACKUP_DIR.joinpath(f"{self.main_dir.name}")
shutil.make_archive(zip_path, "zip", self.main_dir)
shutil.rmtree(TEMP_DIR)
shutil.rmtree(app_dirs.TEMP_DIR)
return str(zip_path.absolute()) + ".zip"
@ -138,10 +138,10 @@ def backup_all(
def auto_backup_job():
for backup in BACKUP_DIR.glob("Auto*.zip"):
for backup in app_dirs.BACKUP_DIR.glob("Auto*.zip"):
backup.unlink()
templates = [template for template in TEMPLATE_DIR.iterdir()]
templates = [template for template in app_dirs.TEMPLATE_DIR.iterdir()]
session = create_session()
backup_all(session=session, tag="Auto", templates=templates)
logger.info("Auto Backup Called")

10
mealie/services/backups/imports.py
View file

@ -4,7 +4,7 @@ import zipfile
from pathlib import Path
from typing import Callable, List
from mealie.core.config import BACKUP_DIR, IMG_DIR, TEMP_DIR
from mealie.core.config import app_dirs
from mealie.db.database import db
from mealie.schema.recipe import Recipe
from mealie.schema.restore import CustomPageImport, GroupImport, RecipeImport, SettingsImport, ThemeImport, UserImport
@ -33,11 +33,11 @@ class ImportDatabase:
Exception: If the zip file does not exists an exception raise.
"""
self.session = session
self.archive = BACKUP_DIR.joinpath(zip_archive)
self.archive = app_dirs.BACKUP_DIR.joinpath(zip_archive)
self.force_imports = force_import
if self.archive.is_file():
self.import_dir = TEMP_DIR.joinpath("active_import")
self.import_dir = app_dirs.TEMP_DIR.joinpath("active_import")
self.import_dir.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(self.archive, "r") as zip_ref:
@ -108,7 +108,7 @@ class ImportDatabase:
image_dir = self.import_dir.joinpath("images")
for image in image_dir.iterdir():
if image.stem in successful_imports:
shutil.copy(image, IMG_DIR)
shutil.copy(image, app_dirs.IMG_DIR)
def import_themes(self):
themes_file = self.import_dir.joinpath("themes", "themes.json")
@ -275,7 +275,7 @@ class ImportDatabase:
return import_status
def clean_up(self):
shutil.rmtree(TEMP_DIR)
shutil.rmtree(app_dirs.TEMP_DIR)
def import_database(

20
mealie/services/image_services.py
View file

@ -2,23 +2,23 @@ import shutil
from pathlib import Path
import requests
from mealie.core.config import IMG_DIR
from fastapi.logger import logger
from mealie.core.config import app_dirs
def read_image(recipe_slug: str) -> Path:
if IMG_DIR.joinpath(recipe_slug).is_file():
return IMG_DIR.joinpath(recipe_slug)
else:
recipe_slug = recipe_slug.split(".")[0]
for file in IMG_DIR.glob(f"{recipe_slug}*"):
return file
if app_dirs.IMG_DIR.joinpath(recipe_slug).is_file():
return app_dirs.IMG_DIR.joinpath(recipe_slug)
recipe_slug = recipe_slug.split(".")[0]
for file in app_dirs.IMG_DIR.glob(f"{recipe_slug}*"):
return file
def write_image(recipe_slug: str, file_data: bytes, extension: str) -> Path.name:
delete_image(recipe_slug)
image_path = Path(IMG_DIR.joinpath(f"{recipe_slug}.{extension}"))
image_path = Path(app_dirs.IMG_DIR.joinpath(f"{recipe_slug}.{extension}"))
with open(image_path, "ab") as f:
f.write(file_data)
@ -27,7 +27,7 @@ def write_image(recipe_slug: str, file_data: bytes, extension: str) -> Path.name
def delete_image(recipe_slug: str) -> str:
recipe_slug = recipe_slug.split(".")[0]
for file in IMG_DIR.glob(f"{recipe_slug}*"):
for file in app_dirs.IMG_DIR.glob(f"{recipe_slug}*"):
return file.unlink()
@ -44,7 +44,7 @@ def scrape_image(image_url: str, slug: str) -> Path:
image_url = image_url.get("url")
filename = slug + "." + image_url.split(".")[-1]
filename = IMG_DIR.joinpath(filename)
filename = app_dirs.IMG_DIR.joinpath(filename)
try:
r = requests.get(image_url, stream=True)

8
mealie/services/migrations/chowdown.py
View file

@ -3,7 +3,7 @@ from pathlib import Path
import yaml
from fastapi.logger import logger
from mealie.core.config import IMG_DIR, TEMP_DIR
from mealie.core.config import app_dirs
from mealie.db.database import db
from mealie.schema.recipe import Recipe
from mealie.utils.unzip import unpack_zip
@ -64,8 +64,8 @@ def chowdown_migrate(session: Session, zip_file: Path):
with temp_dir as dir:
chow_dir = next(Path(dir).iterdir())
image_dir = TEMP_DIR.joinpath(chow_dir, "images")
recipe_dir = TEMP_DIR.joinpath(chow_dir, "_recipes")
image_dir = app_dirs.TEMP_DIR.joinpath(chow_dir, "images")
recipe_dir = app_dirs.TEMP_DIR.joinpath(chow_dir, "_recipes")
failed_recipes = []
successful_recipes = []
@ -83,7 +83,7 @@ def chowdown_migrate(session: Session, zip_file: Path):
for image in image_dir.iterdir():
try:
if image.stem not in failed_recipes:
shutil.copy(image, IMG_DIR.joinpath(image.name))
shutil.copy(image, app_dirs.IMG_DIR.joinpath(image.name))
except Exception as inst:
logger.error(inst)
failed_images.append(image.name)

18
mealie/services/migrations/nextcloud.py
View file

@ -4,10 +4,10 @@ import shutil
import zipfile
from pathlib import Path
from mealie.core.config import IMG_DIR, MIGRATION_DIR, TEMP_DIR
from mealie.core.config import app_dirs
from mealie.db.database import db
from mealie.schema.recipe import Recipe
from mealie.services.scraper.cleaner import Cleaner
from mealie.db.database import db
def process_selection(selection: Path) -> Path:
@ -15,7 +15,7 @@ def process_selection(selection: Path) -> Path:
return selection
elif selection.suffix == ".zip":
with zipfile.ZipFile(selection, "r") as zip_ref:
nextcloud_dir = TEMP_DIR.joinpath("nextcloud")
nextcloud_dir = app_dirs.TEMP_DIR.joinpath("nextcloud")
nextcloud_dir.mkdir(exist_ok=False, parents=True)
zip_ref.extractall(nextcloud_dir)
return nextcloud_dir
@ -46,27 +46,27 @@ def import_recipes(recipe_dir: Path) -> Recipe:
recipe = Recipe(**recipe_data)
if image:
shutil.copy(image, IMG_DIR.joinpath(image_name))
shutil.copy(image, app_dirs.IMG_DIR.joinpath(image_name))
return recipe
def prep():
try:
shutil.rmtree(TEMP_DIR)
shutil.rmtree(app_dirs.TEMP_DIR)
except:
pass
TEMP_DIR.mkdir(exist_ok=True, parents=True)
app_dirs.TEMP_DIR.mkdir(exist_ok=True, parents=True)
def cleanup():
shutil.rmtree(TEMP_DIR)
shutil.rmtree(app_dirs.TEMP_DIR)
def migrate(session, selection: str):
prep()
MIGRATION_DIR.mkdir(exist_ok=True)
selection = MIGRATION_DIR.joinpath(selection)
app_dirs.MIGRATION_DIR.mkdir(exist_ok=True)
selection = app_dirs.MIGRATION_DIR.joinpath(selection)
nextcloud_dir = process_selection(selection)

4
mealie/services/scraper/open_graph.py
View file

@ -1,11 +1,11 @@
from typing import Tuple
import extruct
from mealie.core.config import DEBUG_DIR
from mealie.core.config import app_dirs
from slugify import slugify
from w3lib.html import get_base_url
LAST_JSON = DEBUG_DIR.joinpath("last_recipe.json")
LAST_JSON = app_dirs.DEBUG_DIR.joinpath("last_recipe.json")
def og_field(properties: dict, field_name: str) -> str:

4
mealie/services/scraper/scraper.py
View file

@ -3,14 +3,14 @@ from typing import List
import requests
import scrape_schema_recipe
from mealie.core.config import DEBUG_DIR
from mealie.core.config import app_dirs
from fastapi.logger import logger
from mealie.services.image_services import scrape_image
from mealie.schema.recipe import Recipe
from mealie.services.scraper import open_graph
from mealie.services.scraper.cleaner import Cleaner
LAST_JSON = DEBUG_DIR.joinpath("last_recipe.json")
LAST_JSON = app_dirs.DEBUG_DIR.joinpath("last_recipe.json")
def create_from_url(url: str) -> Recipe:

6
mealie/utils/unzip.py
View file

@ -2,12 +2,12 @@ import tempfile
import zipfile
from pathlib import Path
from mealie.core.config import TEMP_DIR
from mealie.core.config import app_dirs
def unpack_zip(selection: Path) -> tempfile.TemporaryDirectory:
TEMP_DIR.mkdir(parents=True, exist_ok=True)
temp_dir = tempfile.TemporaryDirectory(dir=TEMP_DIR)
app_dirs.TEMP_DIR.mkdir(parents=True, exist_ok=True)
temp_dir = tempfile.TemporaryDirectory(dir=app_dirs.TEMP_DIR)
temp_dir_path = Path(temp_dir.name)
if selection.suffix == ".zip":
with zipfile.ZipFile(selection, "r") as zip_ref:

2
poetry.lock generated
View file

@ -1154,7 +1154,7 @@ python-versions = "*"
[metadata]
lock-version = "1.1"
python-versions = "^3.9"
content-hash = "a95c51999f18bd8ac8e0d08e22d54ef47d00c91ba06e8fbacec2969d466d6cc1"
content-hash = "a6c10e179bc15efc30627c9793218bb944f43dce5e624a7bcabcc47545e661e8"
[metadata.files]
aiofiles = [

1
pyproject.toml
View file

@ -38,6 +38,7 @@ pytest = "^6.2.1"
pytest-cov = "^2.11.0"
mkdocs-material = "^7.0.2"
flake8 = "^3.9.0"
coverage = "^5.5"
[build-system]
requires = ["poetry-core>=1.0.0"]

8
template.env Normal file
View file

@ -0,0 +1,8 @@
DEFAULT_GROUP=Home
ENV=False
API_PORT=9000
API_DOCS=True
DB_TYPE='sqlite'
DEFAULT_PASSWORD=MyPassword
SFTP_USERNAME=None
SFTP_PASSWORD=None

6
tests/conftest.py
View file

@ -3,7 +3,7 @@ import json
import requests
from fastapi.testclient import TestClient
from mealie.app import app
from mealie.core.config import DEFAULT_PASSWORD, SQLITE_DIR
from mealie.core.config import app_dirs, settings
from mealie.db.db_setup import generate_session, sql_global_init
from mealie.db.init_db import init_db
from pytest import fixture
@ -12,7 +12,7 @@ from tests.app_routes import AppRoutes
from tests.test_config import TEST_DATA
from tests.utils.recipe_data import build_recipe_store, get_raw_no_image, get_raw_recipe
SQLITE_FILE = SQLITE_DIR.joinpath("test.db")
SQLITE_FILE = app_dirs.SQLITE_DIR.joinpath("test.db")
SQLITE_FILE.unlink(missing_ok=True)
@ -50,7 +50,7 @@ def test_image():
@fixture(scope="session")
def token(api_client: requests, api_routes: AppRoutes):
form_data = {"username": "changeme@email.com", "password": DEFAULT_PASSWORD}
form_data = {"username": "changeme@email.com", "password": settings.DEFAULT_PASSWORD}
response = api_client.post(api_routes.auth_token, form_data)
token = json.loads(response.text).get("access_token")

10
tests/integration_tests/test_migration_routes.py
View file

@ -4,7 +4,7 @@ from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from mealie.core.config import MIGRATION_DIR
from mealie.core.config import app_dirs
from tests.app_routes import AppRoutes
from tests.test_config import TEST_CHOWDOWN_DIR, TEST_NEXTCLOUD_DIR
@ -29,7 +29,7 @@ def test_upload_chowdown_zip(api_client: TestClient, api_routes: AppRoutes, chow
assert response.status_code == 200
assert MIGRATION_DIR.joinpath("chowdown", chowdown_zip.name).is_file()
assert app_dirs.MIGRATION_DIR.joinpath("chowdown", chowdown_zip.name).is_file()
def test_import_chowdown_directory(api_client: TestClient, api_routes: AppRoutes, chowdown_zip: Path, token):
@ -58,7 +58,7 @@ def test_delete_chowdown_migration_data(api_client: TestClient, api_routes: AppR
response = api_client.delete(delete_url, headers=token)
assert response.status_code == 200
assert not MIGRATION_DIR.joinpath(chowdown_zip.name).is_file()
assert not app_dirs.MIGRATION_DIR.joinpath(chowdown_zip.name).is_file()
# Nextcloud
@ -81,7 +81,7 @@ def test_upload_nextcloud_zip(api_client: TestClient, api_routes: AppRoutes, nex
assert response.status_code == 200
assert MIGRATION_DIR.joinpath("nextcloud", nextcloud_zip.name).is_file()
assert app_dirs.MIGRATION_DIR.joinpath("nextcloud", nextcloud_zip.name).is_file()
def test_import_nextcloud_directory(api_client: TestClient, api_routes: AppRoutes, nextcloud_zip, token):
@ -106,4 +106,4 @@ def test_delete__nextcloud_migration_data(api_client: TestClient, api_routes: Ap
response = api_client.delete(delete_url, headers=token)
assert response.status_code == 200
assert not MIGRATION_DIR.joinpath(nextcloud_zip.name).is_file()
assert not app_dirs.MIGRATION_DIR.joinpath(nextcloud_zip.name).is_file()

84
tests/unit_tests/test_cleaner.py
View file

@ -1,4 +1,46 @@
import json
import re
import pytest
from mealie.services.scraper.cleaner import Cleaner
from mealie.services.scraper.scraper import extract_recipe_from_html
from tests.test_config import TEST_RAW_HTML, TEST_RAW_RECIPES
# https://github.com/django/django/blob/stable/1.3.x/django/core/validators.py#L45
url_validation_regex = re.compile(
r"^(?:http|ftp)s?://" # http:// or https://
r"(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|" # domain...
r"localhost|" # localhost...
r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})" # ...or ip
r"(?::\d+)?" # optional port
r"(?:/?|[/?]\S+)$",
re.IGNORECASE,
)
@pytest.mark.parametrize(
"json_file,num_steps",
[
("best-homemade-salsa-recipe.json", 2),
(
"blue-cheese-stuffed-turkey-meatballs-with-raspberry-balsamic-glaze-2.json",
3,
),
("bon_appetit.json", 8),
("chunky-apple-cake.json", 4),
("dairy-free-impossible-pumpkin-pie.json", 7),
("how-to-make-instant-pot-spaghetti.json", 8),
("instant-pot-chicken-and-potatoes.json", 4),
("instant-pot-kerala-vegetable-stew.json", 13),
("jalapeno-popper-dip.json", 4),
("microwave_sweet_potatoes_04783.json", 4),
("moroccan-skirt-steak-with-roasted-pepper-couscous.json", 4),
("Pizza-Knoblauch-Champignon-Paprika-vegan.html.json", 3),
],
)
def test_cleaner_clean(json_file, num_steps):
recipe_data = Cleaner.clean(json.load(open(TEST_RAW_RECIPES.joinpath(json_file))))
assert len(recipe_data["recipeInstructions"]) == num_steps
def test_clean_category():
@ -7,3 +49,45 @@ def test_clean_category():
def test_clean_html():
assert Cleaner.html("<div>Hello World</div>") == "Hello World"
def test_clean_image():
assert Cleaner.image(None) == "no image"
assert Cleaner.image("https://my.image/path/") == "https://my.image/path/"
assert Cleaner.image({"url": "My URL!"}) == "My URL!"
assert Cleaner.image(["My URL!", "MY SECOND URL"]) == "My URL!"
@pytest.mark.parametrize(
"instructions",
[
"A\n\nB\n\nC\n\n",
"A\nB\nC\n",
"A\r\n\r\nB\r\n\r\nC\r\n\r\n",
"A\r\nB\r\nC\r\n",
["A", "B", "C"],
[{"@type": "HowToStep", "text": x} for x in ["A", "B", "C"]],
],
)
def test_cleaner_instructions(instructions):
assert Cleaner.instructions(instructions) == [
{"text": "A"},
{"text": "B"},
{"text": "C"},
]
def test_html_with_recipe_data():
path = TEST_RAW_HTML.joinpath("healthy_pasta_bake_60759.html")
url = "https://www.bbc.co.uk/food/recipes/healthy_pasta_bake_60759"
recipe_data = extract_recipe_from_html(open(path).read(), url)
assert len(recipe_data["name"]) > 10
assert len(recipe_data["slug"]) > 10
assert recipe_data["orgURL"] == url
assert len(recipe_data["description"]) > 100
assert url_validation_regex.match(recipe_data["image"])
assert len(recipe_data["recipeIngredient"]) == 13
assert len(recipe_data["recipeInstructions"]) == 4

6
tests/unit_tests/test_config.py Normal file
View file

@ -0,0 +1,6 @@
from mealie.core.config import determine_secrets
def test_determine_secret(monkeypatch):
secret = determine_secrets()

7
tests/unit_tests/test_nextcloud.py
View file

@ -1,15 +1,14 @@
from pathlib import Path
import pytest
from mealie.core.config import TEMP_DIR
from mealie.core.config import app_dirs
from mealie.schema.recipe import Recipe
from mealie.services.image_services import IMG_DIR
from mealie.services.migrations.nextcloud import cleanup, import_recipes, prep, process_selection
from tests.test_config import TEST_NEXTCLOUD_DIR
CWD = Path(__file__).parent
TEST_NEXTCLOUD_DIR
TEMP_NEXTCLOUD = TEMP_DIR.joinpath("nextcloud")
TEMP_NEXTCLOUD = app_dirs.TEMP_DIR.joinpath("nextcloud")
@pytest.mark.parametrize(
@ -37,4 +36,4 @@ def test_zip_extraction(file_name: str, final_path: Path):
def test_nextcloud_migration(recipe_dir: Path):
recipe = import_recipes(recipe_dir)
assert isinstance(recipe, Recipe)
IMG_DIR.joinpath(recipe.image).unlink(missing_ok=True)
app_dirs.IMG_DIR.joinpath(recipe.image).unlink(missing_ok=True)

76
tests/unit_tests/test_scraper.py
View file

@ -1,76 +0,0 @@
import json
import re
import pytest
from mealie.services.scraper.cleaner import Cleaner
from mealie.services.scraper.scraper import extract_recipe_from_html
from tests.test_config import TEST_RAW_HTML, TEST_RAW_RECIPES
# https://github.com/django/django/blob/stable/1.3.x/django/core/validators.py#L45
url_validation_regex = re.compile(
r"^(?:http|ftp)s?://" # http:// or https://
r"(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|" # domain...
r"localhost|" # localhost...
r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})" # ...or ip
r"(?::\d+)?" # optional port
r"(?:/?|[/?]\S+)$",
re.IGNORECASE,
)
@pytest.mark.parametrize(
"json_file,num_steps",
[
("best-homemade-salsa-recipe.json", 2),
(
"blue-cheese-stuffed-turkey-meatballs-with-raspberry-balsamic-glaze-2.json",
3,
),
("bon_appetit.json", 8),
("chunky-apple-cake.json", 4),
("dairy-free-impossible-pumpkin-pie.json", 7),
("how-to-make-instant-pot-spaghetti.json", 8),
("instant-pot-chicken-and-potatoes.json", 4),
("instant-pot-kerala-vegetable-stew.json", 13),
("jalapeno-popper-dip.json", 4),
("microwave_sweet_potatoes_04783.json", 4),
("moroccan-skirt-steak-with-roasted-pepper-couscous.json", 4),
("Pizza-Knoblauch-Champignon-Paprika-vegan.html.json", 3),
],
)
def test_normalize_data(json_file, num_steps):
recipe_data = Cleaner.clean(json.load(open(TEST_RAW_RECIPES.joinpath(json_file))))
assert len(recipe_data["recipeInstructions"]) == num_steps
@pytest.mark.parametrize(
"instructions",
[
"A\n\nB\n\nC\n\n",
"A\nB\nC\n",
"A\r\n\r\nB\r\n\r\nC\r\n\r\n",
"A\r\nB\r\nC\r\n",
["A", "B", "C"],
[{"@type": "HowToStep", "text": x} for x in ["A", "B", "C"]],
],
)
def test_normalize_instructions(instructions):
assert Cleaner.instructions(instructions) == [
{"text": "A"},
{"text": "B"},
{"text": "C"},
]
def test_html_with_recipe_data():
path = TEST_RAW_HTML.joinpath("healthy_pasta_bake_60759.html")
url = "https://www.bbc.co.uk/food/recipes/healthy_pasta_bake_60759"
recipe_data = extract_recipe_from_html(open(path).read(), url)
assert len(recipe_data["name"]) > 10
assert len(recipe_data["slug"]) > 10
assert recipe_data["orgURL"] == url
assert len(recipe_data["description"]) > 100
assert url_validation_regex.match(recipe_data["image"])
assert len(recipe_data["recipeIngredient"]) == 13
assert len(recipe_data["recipeInstructions"]) == 4