Skip to content

Manifest

application_kit.settings

get_redis_instance_and_extras 

get_redis_instance_and_extras(dependency_name)

Returns a redis instance for the dependency name.

Source code in application_kit/settings.py 
27
28
29
30
31
32
33
34
35
36
37
38
def get_redis_instance_and_extras(
    dependency_name: str,
) -> tuple[redis.Redis[bytes] | None, dict[str, Any] | None]:
    """Returns a redis instance for the dependency name."""
    redis_instance, extra_parameters = get_from_redis_cache(dependency_name)
    if redis_instance is None:
        redis_args = get_redis_args_for_dependency(dependency_name)
        redis_instance, extra_parameters = _get_redis_instance(**redis_args)
        Pin.override(redis_instance, service=f"redis.{dependency_name}")
        add_to_redis_cache(dependency_name, (redis_instance, extra_parameters))

    return redis_instance, extra_parameters

get_redis_instance 

get_redis_instance(dependency_name)

Returns a redis instance from the dependency named dependency_name

Source code in application_kit/settings.py 
55
56
57
58
59
60
def get_redis_instance(dependency_name: str) -> redis.Redis[bytes]:
    """Returns a redis instance from the dependency named dependency_name"""
    redis_instance, _ = get_redis_instance_and_extras(dependency_name)
    if redis_instance is None:
        raise RedisError(f"{dependency_name} was not fulfilled.")
    return redis_instance

get_configuration 

get_configuration(configuration_name)

Returns the configuration value.

Source code in application_kit/settings.py 
63
64
65
def get_configuration(configuration_name: str) -> Any:
    """Returns the configuration value."""
    return get_bender_settings().get(configuration_name)

get_service_url 

get_service_url(service_name, path=None)

Returns the service URL for service_name.

Source code in application_kit/settings.py 
68
69
70
71
72
73
def get_service_url(service_name: str, path: str | None = None) -> str | None:
    """Returns the service URL for service_name."""
    if path is not None:
        return get_bender_settings().get_service_url(service_name, path)

    return get_bender_settings().get_service(service_name)

get_django_databases 

get_django_databases()

Returns a django database dict.

Source code in application_kit/settings.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def get_django_databases() -> dict[str, dict[str, Any]]:
    """Returns a django database dict."""
    bender_settings = get_bender_settings()
    databases = bender_settings.filter_database_dependencies(DatabaseType.postgresql)
    retval = {}

    for name, database_dependency in databases.items():
        env_key = database_dependency.config_key()
        url = BenderDbUrl(bender_settings.configuration[env_key])

        if "postgis" in database_dependency.extensions:
            engine = "django.contrib.gis.db.backends.postgis"
        else:
            engine = "django.db.backends.postgresql"

        database_config = _make_django_db_conf(engine, url)

        retval[name] = database_config

    return retval

get_database_url 

get_database_url(name)

Returns a BenderDbUrl for the database dependency named name.

Source code in application_kit/settings.py 
 98
 99
100
101
def get_database_url(name: str) -> BenderDbUrl | None:
    """Returns a BenderDbUrl for the database dependency named name."""
    bender_settings = get_bender_settings()
    return bender_settings.get_database(name)

get_version 

get_version()

Returns the current version.

Source code in application_kit/settings.py 
117
118
119
120
121
def get_version() -> str:
    """Returns the current version."""
    bender_settings = get_bender_settings()
    configuration_backend = bender_settings.configuration
    return configuration_backend.fallback("VERSION", "DD_VERSION", default="local_dev")

get_environment 

get_environment()

Returns the current Environment.

Source code in application_kit/settings.py 
124
125
126
127
128
def get_environment() -> str | None:
    """Returns the current Environment."""
    bender_settings = get_bender_settings()
    configuration_backend = bender_settings.configuration
    return configuration_backend.fallback("ENVIRONMENT", "DATADOG_ENV")

get_service_name 

get_service_name(worker_name)

Returns the current service name, or optionally worker name.

Source code in application_kit/settings.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def get_service_name(worker_name: str | None) -> str:
    """Returns the current service name, or optionally worker name."""
    bender_settings = get_bender_settings()

    configuration_backend = bender_settings.configuration
    application_manifest = bender_settings.manifest

    override_service_name = configuration_backend.fallback("DATADOG_SERVICE", "DD_SERVICE")

    if override_service_name is None:
        application_name = application_manifest.name
        service_name = configuration_backend.get("SERVICE_NAME")

        if worker_name is not None and worker_name not in application_manifest.workers.keys():
            raise ConfigurationError(f"{worker_name} is not defined in application manifest workers dictionary.")

        if service_name is not None and service_name not in application_manifest.http.keys():
            raise ConfigurationError(f"{service_name} is not defined in application manifest http dictionary.")

        worker_name = worker_name or service_name
        if worker_name is not None:
            service_name = "_".join([application_name, worker_name])
        else:
            service_name = application_name

    else:
        service_name = override_service_name

    return service_name

get_async_redis_instance_and_extras 

get_async_redis_instance_and_extras(dependency_name)

Returns a redis instance for the dependency name.

Source code in application_kit/settings.py 
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def get_async_redis_instance_and_extras(
    dependency_name: str,
) -> tuple[redis.asyncio.Redis[bytes] | None, dict[str, Any] | None]:
    """Returns a redis instance for the dependency name."""
    connection_pool, extra_parameters = get_from_async_redis_cache(dependency_name)
    if connection_pool is None:
        redis_args = get_redis_args_for_dependency(dependency_name)
        connection_pool, extra_parameters = _get_async_redis_connection_pool(**redis_args)
        add_to_async_redis_cache(dependency_name, (connection_pool, extra_parameters))

    redis_instance = redis.asyncio.Redis(connection_pool=connection_pool)
    Pin.override(redis_instance, service=f"async_redis.{dependency_name}")

    return redis_instance, extra_parameters