App Settings¶
There are various ways to configure your application. Recommended way is to use the setup hook to benefit Rocketry’s dynamic arguments.
Here is a simple example to set some configurations:
from rocketry import Rocketry
from rocketry.args import Config
app = Rocketry()
@app.setup()
def set_config(config=Config()):
config.silence_task_prerun = True
config.silence_task_logging = True
config.silence_cond_check = True
The argument config
will get app.session.config
as the value. This function will run before any tasks
are run. Therefore it is safe to set the log record repository
or even dynamically create tasks in this function.
Note
You can split your setup to multiple functions and use
@app.setup()
multiple times.
Here is a more sophisticated example where we log to an SQL database and change the configurations based on whether we are in production or test:
from redbird.repos import SQLRepo
from sqlalchemy import create_engine
from rocketry import Rocketry
from rocketry.args import TaskLogger, Config, EnvArg, Session
from rocketry.log import MinimalRecord
app = Rocketry()
@app.setup()
def set_repo(logger=TaskLogger()):
repo = SQLRepo(engine=create_engine("sqlite:///app.db"), table="tasks", model=MinimalRecord, id_field="created")
logger.set_repo(repo)
@app.setup()
def set_config(config=Config(), env=EnvArg("ENV", default="dev")):
if env == "prod":
config.silence_task_prerun = True
config.silence_task_logging = True
config.silence_cond_check = True
else:
config.silence_task_prerun = False
config.silence_task_logging = False
config.silence_cond_check = False
Note
The tasks’ caches (ie. status and last run/success/fail) are set after the hooks have run. If your setup needs to run after the caches are set and startup tasks have run, you can do it by:
@app.setup()
def setup_app():
# Run before startup tasks and cache is set
...
yield
# Run after startup tasks and cache is set
...
You can also modify tasks in the setup. For example, if you wish to have an environment to test only the scheduling (without running anything):
def do_nothing(*args, **kwargs):
...
@app.setup()
def set_tasks(session=Session(), env=EnvArg("ENV", default="dev")):
if env == "test-schedule":
# Set all tasks to run nothing
# to test the scheduling works
for task in session.tasks:
task.func = do_nothing