fastapi_celery/tasks.py

"""
A set of `invoke` helper commands used for this project.
"""
from invoke import task


@task
def start_app(c):
    """Starts the app."""
    print("Starting the developer environment...")
    start_redis(c)
    c.run("uvicorn main:app --reload", pty=True)


@task
def run_unit_tests(c):
    """Triggers a local unittest run for the app."""
    # Pytest will hide color output when it thinks it's running
    # outside of a terminal therefore pty=True is used.
    c.run("python3 -m pytest -vvv tests/", pty=True)


@task
def fix_eof(c):
    """Fixes any missing newlines for end of files."""
    c.run("pre-commit run end-of-file-fixer --all-files", pty=True)


@task
def start_redis(c):
    """Runs the Redis integration environment."""
    print("Starting Redis")
    c.run("docker compose -f docker-compose-redis.yml up -d")


@task
def stop_redis(c):
    """Stops the Redis integration environent."""
    print("Stopping Redis")
    c.run("docker compose -f docker-compose-redis.yml down")


@task
def create_migration(c):
    """Creates a sqlalchemy migration."""
    c.run("alembic revision --autogenerate")


@task
def apply_migration(c, revision="head"):
    """Applies an Alembic migration."""
    c.run(f"alembic upgrade {revision}")


@task(help={"username": "Username of the application", "email": "User's email"})
def add_user(username="", email=""):
    "Adds a dummy user to the application Database"
    if not username or email:
        raise ValueError("You must provide a username and email")
    from project.database import SessionLocal
    from project.users.models import User

    from main import app  # pylint: disable=unused-import

    session = SessionLocal()
    user = User(username=username, email=email)
    session.add(user)
    session.commit()
    session.close()