You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
fastapi_celery/tasks.py

91 lines
2.3 KiB
Python

"""
A set of `invoke` helper commands used for this project.
"""
from invoke import task
# pylint: disable=unused-argument
@task
def start_dev(c):
"""Starts the app in a docker-compose developer environment."""
print("Starting the developer environment...")
c.run("docker-compose -f docker-compose.yml up -d")
@task
def stop_dev(c):
"""Stops the developer environment."""
print("Stoping local development environment.")
c.run("docker-compose -f docker-compose.yml down")
@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": "A username", "email": "The user's email"})
def add_user(c, 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()
@task(
help={
"requirements": "Path to a pip requirements file.",
"dev_requirements": "Path to a pip dev requirements file.",
}
)
def sync_env(
c, requirements="requirements.txt", dev_requirements="dev-requirements.txt"
):
"""Uses pip-sync to sync the local virtualenv."""
c.run(f"pip-sync {requirements} {dev_requirements}")