# -*- coding: utf-8 -*- # type: ignore from __future__ import annotations import functools import os from invoke import task from invoke.exceptions import UnexpectedExit from hotpocket_workspace_tools import WorkspaceMode, get_workspace_mode WORKSPACE_MODE = get_workspace_mode() ENV = os.getenv('HOTPOCKET_BACKEND_ENV', 'docker') @task def clean(ctx): ctx.run('rm -rf hotpocket_backend/static/') @task def test(ctx): ctx.run('pytest -v --disable-warnings') @task def flake8(ctx): ctx.run('flake8') @task def isort(ctx, check=False, diff=False): command_parts = [ 'isort', ] if check is True: command_parts.append('--check') if diff is True: command_parts.append('--diff') command_parts.append('.') ctx.run(' '.join(command_parts)) @task def eslint(ctx): ctx.run('npx eslint') @task def lint(ctx): ihazsuccess = True try: flake8(ctx) except UnexpectedExit: ihazsuccess = False try: isort(ctx, check=True) except UnexpectedExit: ihazsuccess = False try: eslint(ctx) except UnexpectedExit: ihazsuccess = False if ihazsuccess is False: raise RuntimeError('FIAL') @task def typecheck(ctx): ctx.run('mypy .') @task def django_shell(ctx): ctx.run('python manage.py shell_plus') @task def ci(ctx): ihazsuccess = True ci_tasks = [ test, flake8, functools.partial(isort, check=True), typecheck, ] for ci_task in ci_tasks: try: ci_task(ctx) except UnexpectedExit: ihazsuccess = False if ihazsuccess is False: raise RuntimeError('FIAL') @task def setup(ctx): ctx.run('python manage.py migrate') ctx.run('python manage.py create_initial_account hotpocket hotpocketm4st3r') if WORKSPACE_MODE == WorkspaceMode.METAL: ctx.run('mkdir -p run/uploads') @task def start_web(ctx): bind = os.getenv('HOTPOCKET_BACKEND_RUNSERVER_BIND_WEB', '127.0.0.1:8001') ctx.run(f'python manage.py runserver {bind}') @task def start_admin(ctx): bind = os.getenv('HOTPOCKET_BACKEND_RUNSERVER_BIND_ADMIN', '127.0.0.1:8002') ctx.run(f'python manage.py runserver {bind}', env=dict( DJANGO_SETTINGS_MODULE=f'hotpocket_backend.settings.{ENV}.admin', HOTPOCKET_BACKEND_APP='admin', )) @task def start_celery_worker(ctx, concurrency=2): command_parts = [ 'celery', '-A hotpocket_backend.celery:app', 'worker', '-l INFO', '-Q celery,webapp', f'-c {concurrency}', ] ctx.run(' '.join(command_parts)) @task def start_celery_beat(ctx, schedule_path='run/celery-beat-schedule'): ctx.run(f'celery -A hotpocket_backend.celery:app beat -l INFO -s {schedule_path}')