Skip to content

Index

The main entrypoint for GSO, and the different ways in which it can be run.

gso_initialise_celery(celery)

Initialise the Celery app.

Source code in gso/__init__.py 
24
25
26
27
def gso_initialise_celery(celery: Celery) -> None:
    """Initialise the Celery app."""
    initialise_celery(celery)
    celery.conf.task_routes = {}

init_gso_app()

Initialise the GSO app.

Source code in gso/__init__.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def init_gso_app() -> OrchestratorCore:
    """Initialise the GSO app."""
    app = OrchestratorCore(base_settings=app_settings)
    app.register_authentication(oidc_instance)
    app.register_authorization(opa_instance)
    app.register_graphql_authorization(graphql_opa_instance)
    app.register_graphql(subscription_interface=custom_subscription_interface)
    app.include_router(api_router, prefix="/api")

    celery = Celery(
        "geant-service-orchestrator",
        broker=celery_settings.broker_url,
        backend=celery_settings.result_backend,
        include=["orchestrator.services.tasks", "gso.tasks.start_process"],
    )
    celery.conf.update(
        result_expires=celery_settings.result_expires,
        task_always_eager=app_settings.TESTING,
        task_eager_propagates=app_settings.TESTING,
    )
    gso_initialise_celery(celery)

    return app

init_cli_app()

Initialise GSO as a CLI application.

Source code in gso/__init__.py 
55
56
57
58
59
60
61
def init_cli_app() -> typer.Typer:
    """Initialise GSO as a CLI application."""
    from gso.cli import imports, netbox  # noqa: PLC0415

    cli_app.add_typer(imports.app, name="import-cli")
    cli_app.add_typer(netbox.app, name="netbox-cli")
    return cli_app()

init_sentry()

Only initialize Sentry if not in testing mode.

Source code in gso/__init__.py 
64
65
66
67
68
69
def init_sentry() -> None:
    """Only initialize Sentry if not in testing mode."""
    if not app_settings.TESTING and (sentry_params := load_oss_params().SENTRY):
        sentry_sdk.init(
            dsn=sentry_params.DSN, environment=load_oss_params().GENERAL.environment, traces_sample_rate=1.0
        )