Skip to content

Core

tangram_core

InjectBackendState module-attribute

BackendState dataclass

Source code in packages/tangram_core/src/tangram_core/backend.py
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@dataclass
class BackendState:
    redis_client: redis.Redis
    http_client: httpx.AsyncClient
    config: Config
    loaded_plugins: dict[str, Plugin] = field(default_factory=dict)

    @property
    def base_url(self) -> str:
        host = self.config.server.host
        port = self.config.server.port
        if host == "0.0.0.0":
            host = "127.0.0.1"
        return f"http://{host}:{port}"

redis_client instance-attribute

redis_client: Redis

http_client instance-attribute

http_client: AsyncClient

config instance-attribute

config: Config

loaded_plugins class-attribute instance-attribute

loaded_plugins: dict[str, Plugin] = field(default_factory=dict)

base_url property

base_url: str

Runtime

Manages the lifecycle of the Tangram backend, including the Uvicorn server, background services, and connection pools (Redis, HTTPX).

Source code in packages/tangram_core/src/tangram_core/backend.py
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
class Runtime:
    """Manages the lifecycle of the Tangram backend, including the
    Uvicorn server, background services, and connection pools (Redis, HTTPX).
    """

    def __init__(self, config: IntoConfig | None = None) -> None:
        if isinstance(config, (str, Path, os.PathLike)):
            self.config = Config.from_file(config)
        else:
            self.config = config or Config()
        self._stack = AsyncExitStack()
        self._state: BackendState | None = None
        self._server: uvicorn.Server | None = None
        self._server_task: asyncio.Task[None] | None = None
        self._service_tasks: list[asyncio.Task[None]] = []

    @property
    def state(self) -> BackendState:
        if self._state is None:
            raise RuntimeError("runtime is not started, call start() first")
        return self._state

    async def start(self) -> Self:
        """Starts the backend runtime."""
        if self._state is not None:
            raise RuntimeError("runtime is already started")

        redis_client = await self._stack.enter_async_context(
            redis.from_url(self.config.core.redis_url)  # type: ignore
        )
        http_client = await self._stack.enter_async_context(
            httpx.AsyncClient(http2=True)
        )
        self._state = BackendState(
            redis_client=redis_client,
            http_client=http_client,
            config=self.config,
        )

        loaded_plugins = load_enabled_plugins(self.config)
        app = create_app(self._state, loaded_plugins)

        server_config = uvicorn.Config(
            app,
            host=self.config.server.host,
            port=self.config.server.port,
            log_config=get_log_config_dict(self.config),
        )
        self._server = uvicorn.Server(server_config)

        self._service_tasks.append(
            asyncio.create_task(run_channel_service(self.config))
        )
        for plugin in loaded_plugins:
            for _, service_func in sorted(
                plugin.services, key=lambda s: (s[0], s[1].__name__)
            ):
                self._service_tasks.append(
                    asyncio.create_task(service_func(self._state))
                )
                logger.info(f"started service from plugin: {plugin.dist_name}")

        self._server_task = asyncio.create_task(self._server.serve())

        while not self._server.started:
            if self._server_task.done():
                await self._server_task
            await asyncio.sleep(0.1)

        return self

    async def wait(self) -> None:
        """Waits for the server task to complete (e.g. via signal or internal error)."""
        if self._server_task:
            try:
                await self._server_task
            except asyncio.CancelledError:
                pass

    async def stop(self) -> None:
        """Stops the backend runtime."""
        if self._server and self._server.started:
            self._server.should_exit = True
            if self._server_task:
                try:
                    await self._server_task
                except asyncio.CancelledError:
                    pass

        for task in self._service_tasks:
            task.cancel()
        if self._service_tasks:
            await asyncio.gather(*self._service_tasks, return_exceptions=True)
        self._service_tasks.clear()

        await self._stack.aclose()
        self._state = None
        self._server = None
        self._server_task = None

    async def __aenter__(self) -> Self:
        return await self.start()

    async def __aexit__(self, *args: object) -> None:
        await self.stop()

config instance-attribute

config = Config.from_file(config)

state property

state: BackendState

start async

start() -> Self

Starts the backend runtime.

Source code in packages/tangram_core/src/tangram_core/backend.py
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
async def start(self) -> Self:
    """Starts the backend runtime."""
    if self._state is not None:
        raise RuntimeError("runtime is already started")

    redis_client = await self._stack.enter_async_context(
        redis.from_url(self.config.core.redis_url)  # type: ignore
    )
    http_client = await self._stack.enter_async_context(
        httpx.AsyncClient(http2=True)
    )
    self._state = BackendState(
        redis_client=redis_client,
        http_client=http_client,
        config=self.config,
    )

    loaded_plugins = load_enabled_plugins(self.config)
    app = create_app(self._state, loaded_plugins)

    server_config = uvicorn.Config(
        app,
        host=self.config.server.host,
        port=self.config.server.port,
        log_config=get_log_config_dict(self.config),
    )
    self._server = uvicorn.Server(server_config)

    self._service_tasks.append(
        asyncio.create_task(run_channel_service(self.config))
    )
    for plugin in loaded_plugins:
        for _, service_func in sorted(
            plugin.services, key=lambda s: (s[0], s[1].__name__)
        ):
            self._service_tasks.append(
                asyncio.create_task(service_func(self._state))
            )
            logger.info(f"started service from plugin: {plugin.dist_name}")

    self._server_task = asyncio.create_task(self._server.serve())

    while not self._server.started:
        if self._server_task.done():
            await self._server_task
        await asyncio.sleep(0.1)

    return self

wait async

wait() -> None

Waits for the server task to complete (e.g. via signal or internal error).

Source code in packages/tangram_core/src/tangram_core/backend.py
497
498
499
500
501
502
503
async def wait(self) -> None:
    """Waits for the server task to complete (e.g. via signal or internal error)."""
    if self._server_task:
        try:
            await self._server_task
        except asyncio.CancelledError:
            pass

stop async

stop() -> None

Stops the backend runtime.

Source code in packages/tangram_core/src/tangram_core/backend.py
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
async def stop(self) -> None:
    """Stops the backend runtime."""
    if self._server and self._server.started:
        self._server.should_exit = True
        if self._server_task:
            try:
                await self._server_task
            except asyncio.CancelledError:
                pass

    for task in self._service_tasks:
        task.cancel()
    if self._service_tasks:
        await asyncio.gather(*self._service_tasks, return_exceptions=True)
    self._service_tasks.clear()

    await self._stack.aclose()
    self._state = None
    self._server = None
    self._server_task = None

Config dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
@dataclass
class Config:
    core: CoreConfig = field(default_factory=CoreConfig)
    server: ServerConfig = field(default_factory=ServerConfig)
    channel: ChannelConfig = field(default_factory=ChannelConfig)
    map: MapConfig = field(default_factory=MapConfig)
    plugins: dict[str, Any] = field(default_factory=dict)
    """Mapping of plugin name to plugin-specific config."""
    cache: CacheConfig = field(default_factory=CacheConfig)

    @classmethod
    def from_file(cls, config_path: StrOrPathLike) -> Config:
        if sys.version_info < (3, 11):
            import tomli as tomllib
        else:
            import tomllib
        from pydantic import TypeAdapter

        with open(config_path, "rb") as f:
            cfg_data = tomllib.load(f)

        base_dir = Path(config_path).parent
        map = cfg_data.setdefault("map", {})
        if (s := map.get("style", None)) is not None:
            map["style"] = try_resolve_local_style(base_dir, s)

        if "styles" in map:
            map["styles"] = [
                try_resolve_local_style(base_dir, style)
                for style in map.get("styles", [])
            ]

        config_adapter = TypeAdapter(cls)
        config = config_adapter.validate_python(cfg_data)
        return config

core class-attribute instance-attribute

core: CoreConfig = field(default_factory=CoreConfig)

server class-attribute instance-attribute

server: ServerConfig = field(default_factory=ServerConfig)

channel class-attribute instance-attribute

channel: ChannelConfig = field(default_factory=ChannelConfig)

map class-attribute instance-attribute

map: MapConfig = field(default_factory=MapConfig)

plugins class-attribute instance-attribute

plugins: dict[str, Any] = field(default_factory=dict)

Mapping of plugin name to plugin-specific config.

cache class-attribute instance-attribute

cache: CacheConfig = field(default_factory=CacheConfig)

from_file classmethod

from_file(config_path: StrOrPathLike) -> Config
Source code in packages/tangram_core/src/tangram_core/config.py
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
@classmethod
def from_file(cls, config_path: StrOrPathLike) -> Config:
    if sys.version_info < (3, 11):
        import tomli as tomllib
    else:
        import tomllib
    from pydantic import TypeAdapter

    with open(config_path, "rb") as f:
        cfg_data = tomllib.load(f)

    base_dir = Path(config_path).parent
    map = cfg_data.setdefault("map", {})
    if (s := map.get("style", None)) is not None:
        map["style"] = try_resolve_local_style(base_dir, s)

    if "styles" in map:
        map["styles"] = [
            try_resolve_local_style(base_dir, style)
            for style in map.get("styles", [])
        ]

    config_adapter = TypeAdapter(cls)
    config = config_adapter.validate_python(cfg_data)
    return config

Plugin dataclass

Stores the metadata and registered API routes, background services and frontend assets for a tangram plugin.

Packages should declare an entry point in the tangram_core.plugins group in their pyproject.toml pointing to an instance of this class.

Source code in packages/tangram_core/src/tangram_core/plugin.py
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@dataclass(eq=False)
class Plugin:
    """Stores the metadata and registered API routes, background services and
    frontend assets for a tangram plugin.

    Packages should declare an entry point in the `tangram_core.plugins` group
    in their `pyproject.toml` pointing to an instance of this class.
    """

    frontend_path: str | None = None
    """Path to the compiled frontend assets, *relative* to the distribution root
    (editable) or package root (wheel).
    """
    routers: list[APIRouter] = field(default_factory=list)
    config_class: type | None = None
    """The **backend** configuration class (dataclass, TypedDict or Pydantic model)
    for the plugin."""
    frontend_config_class: type | None = None
    """The **frontend** configuration class for the plugin. If set, it will be used to
    generate the frontend schema and validate settings updates.
    Fields should be annotated with [tangram_core.config.FrontendMutable][] if they
    are allowed to be modified from the frontend settings UI.
    """
    into_frontend_config_function: IntoFrontendConfigFunction | None = None
    """Function to transform the backend configuration into the frontend
    configuration. It receives the validated backend configuration object and
    should return an instance of `frontend_config_class`.
    Useful if the user wants to hide sensitive fields (e.g. API keys) from the frontend
    or dynamically compute certain fields.
    Required if `frontend_config_class` is set.
    """
    lifespan: Lifespan | None = None
    """Async context manager for plugin initialization and teardown."""
    services: list[tuple[Priority, ServiceAsyncFunc]] = field(
        default_factory=list, init=False
    )
    get_typer: Callable[[], Typer] | None = None
    """A function that returns the subcommands which will later be registered."""
    dist_name: str = field(init=False)
    """Name of the distribution (package) that provided this plugin, populated
    automatically during loading.
    """  # we do this so plugins can know their own package name if needed

    def register_service(
        self, priority: Priority = 0
    ) -> Callable[[ServiceFunc], ServiceFunc]:
        """Decorator to register a background service function.

        Services are long-running async functions that receive the BackendState
        and are started when the application launches.
        """

        def decorator(func: ServiceFunc) -> ServiceFunc:
            @functools.wraps(func)
            async def async_wrapper(backend_state: BackendState) -> None:
                if asyncio.iscoroutinefunction(func):
                    await func(backend_state)
                else:
                    await asyncio.to_thread(func, backend_state)

            self.services.append((priority, async_wrapper))
            return func

        return decorator

    # keeping the public methods callable for plugin API compatibility.
    # plugins are mutable registries, so equality and hashing use object identity
    # method-level lru_cache would hash and retain mutable plugins

    @functools.cached_property
    def _adapter(self) -> TypeAdapter[Any] | None:
        from pydantic import TypeAdapter

        if self.config_class:
            return TypeAdapter(self.config_class)
        return None

    def adapter(self) -> TypeAdapter[Any] | None:
        """Returns a cached Pydantic TypeAdapter for the plugin's configuration class.

        Avoids expensive rebuilds on every validation request, such as those from the
        settings UI.
        """
        return self._adapter

    @functools.cached_property
    def _frontend_adapter(self) -> TypeAdapter[Any] | None:
        from pydantic import TypeAdapter

        if self.frontend_config_class:
            return TypeAdapter(self.frontend_config_class)
        return None

    def frontend_adapter(self) -> TypeAdapter[Any] | None:
        """Returns a cached Pydantic TypeAdapter for the plugin's frontend
        configuration class."""
        return self._frontend_adapter

frontend_path class-attribute instance-attribute

frontend_path: str | None = None

Path to the compiled frontend assets, relative to the distribution root (editable) or package root (wheel).

routers class-attribute instance-attribute

routers: list[APIRouter] = field(default_factory=list)

config_class class-attribute instance-attribute

config_class: type | None = None

The backend configuration class (dataclass, TypedDict or Pydantic model) for the plugin.

frontend_config_class class-attribute instance-attribute

frontend_config_class: type | None = None

The frontend configuration class for the plugin. If set, it will be used to generate the frontend schema and validate settings updates. Fields should be annotated with tangram_core.config.FrontendMutable if they are allowed to be modified from the frontend settings UI.

into_frontend_config_function class-attribute instance-attribute

into_frontend_config_function: IntoFrontendConfigFunction | None = None

Function to transform the backend configuration into the frontend configuration. It receives the validated backend configuration object and should return an instance of frontend_config_class. Useful if the user wants to hide sensitive fields (e.g. API keys) from the frontend or dynamically compute certain fields. Required if frontend_config_class is set.

lifespan class-attribute instance-attribute

lifespan: Lifespan | None = None

Async context manager for plugin initialization and teardown.

services class-attribute instance-attribute

services: list[tuple[Priority, ServiceAsyncFunc]] = field(default_factory=list, init=False)

get_typer class-attribute instance-attribute

get_typer: Callable[[], Typer] | None = None

A function that returns the subcommands which will later be registered.

dist_name class-attribute instance-attribute

dist_name: str = field(init=False)

Name of the distribution (package) that provided this plugin, populated automatically during loading.

register_service

register_service(priority: Priority = 0) -> Callable[[ServiceFunc], ServiceFunc]

Decorator to register a background service function.

Services are long-running async functions that receive the BackendState and are started when the application launches.

Source code in packages/tangram_core/src/tangram_core/plugin.py
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def register_service(
    self, priority: Priority = 0
) -> Callable[[ServiceFunc], ServiceFunc]:
    """Decorator to register a background service function.

    Services are long-running async functions that receive the BackendState
    and are started when the application launches.
    """

    def decorator(func: ServiceFunc) -> ServiceFunc:
        @functools.wraps(func)
        async def async_wrapper(backend_state: BackendState) -> None:
            if asyncio.iscoroutinefunction(func):
                await func(backend_state)
            else:
                await asyncio.to_thread(func, backend_state)

        self.services.append((priority, async_wrapper))
        return func

    return decorator

adapter

adapter() -> TypeAdapter[Any] | None

Returns a cached Pydantic TypeAdapter for the plugin's configuration class.

Avoids expensive rebuilds on every validation request, such as those from the settings UI.

Source code in packages/tangram_core/src/tangram_core/plugin.py
109
110
111
112
113
114
115
def adapter(self) -> TypeAdapter[Any] | None:
    """Returns a cached Pydantic TypeAdapter for the plugin's configuration class.

    Avoids expensive rebuilds on every validation request, such as those from the
    settings UI.
    """
    return self._adapter

frontend_adapter

frontend_adapter() -> TypeAdapter[Any] | None

Returns a cached Pydantic TypeAdapter for the plugin's frontend configuration class.

Source code in packages/tangram_core/src/tangram_core/plugin.py
125
126
127
128
def frontend_adapter(self) -> TypeAdapter[Any] | None:
    """Returns a cached Pydantic TypeAdapter for the plugin's frontend
    configuration class."""
    return self._frontend_adapter

backend

logger module-attribute

logger = logging.getLogger(__name__)

InjectBackendState module-attribute

CACHE_PARAM_PATTERN module-attribute

CACHE_PARAM_PATTERN = re.compile('\\{(\\w+)\\}')

DEFAULT_THEMES module-attribute

DEFAULT_THEMES = (ThemeDefinition(name='light', background='#ffffff', foreground='#000000', surface='#f8f9fa', border='#e7e7e7', hover='#e9ecef', accent1='oklch(0.5616 0.0895 251.64)', accent1_foreground='#ffffff', accent2='oklch(0.8021 0.11 92.43)', accent2_foreground='#000000', muted='#666666', error='#8e1b27'), ThemeDefinition(name='dark', background='#1a1a1a', foreground='#e0e0e0', surface='#2d2d2d', border='#404040', hover='#343434', accent1='oklch(0.5059 0.0895 251.64)', accent1_foreground='#ffffff', accent2='oklch(0.5059 0.0895 93.53)', accent2_foreground='#ffffff', muted='#999999', error='#844d53'))

BackendState dataclass

Source code in packages/tangram_core/src/tangram_core/backend.py
58
59
60
61
62
63
64
65
66
67
68
69
70
71
@dataclass
class BackendState:
    redis_client: redis.Redis
    http_client: httpx.AsyncClient
    config: Config
    loaded_plugins: dict[str, Plugin] = field(default_factory=dict)

    @property
    def base_url(self) -> str:
        host = self.config.server.host
        port = self.config.server.port
        if host == "0.0.0.0":
            host = "127.0.0.1"
        return f"http://{host}:{port}"
redis_client instance-attribute
redis_client: Redis
http_client instance-attribute
http_client: AsyncClient
config instance-attribute
config: Config
loaded_plugins class-attribute instance-attribute
loaded_plugins: dict[str, Plugin] = field(default_factory=dict)
base_url property
base_url: str

Runtime

Manages the lifecycle of the Tangram backend, including the Uvicorn server, background services, and connection pools (Redis, HTTPX).

Source code in packages/tangram_core/src/tangram_core/backend.py
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
class Runtime:
    """Manages the lifecycle of the Tangram backend, including the
    Uvicorn server, background services, and connection pools (Redis, HTTPX).
    """

    def __init__(self, config: IntoConfig | None = None) -> None:
        if isinstance(config, (str, Path, os.PathLike)):
            self.config = Config.from_file(config)
        else:
            self.config = config or Config()
        self._stack = AsyncExitStack()
        self._state: BackendState | None = None
        self._server: uvicorn.Server | None = None
        self._server_task: asyncio.Task[None] | None = None
        self._service_tasks: list[asyncio.Task[None]] = []

    @property
    def state(self) -> BackendState:
        if self._state is None:
            raise RuntimeError("runtime is not started, call start() first")
        return self._state

    async def start(self) -> Self:
        """Starts the backend runtime."""
        if self._state is not None:
            raise RuntimeError("runtime is already started")

        redis_client = await self._stack.enter_async_context(
            redis.from_url(self.config.core.redis_url)  # type: ignore
        )
        http_client = await self._stack.enter_async_context(
            httpx.AsyncClient(http2=True)
        )
        self._state = BackendState(
            redis_client=redis_client,
            http_client=http_client,
            config=self.config,
        )

        loaded_plugins = load_enabled_plugins(self.config)
        app = create_app(self._state, loaded_plugins)

        server_config = uvicorn.Config(
            app,
            host=self.config.server.host,
            port=self.config.server.port,
            log_config=get_log_config_dict(self.config),
        )
        self._server = uvicorn.Server(server_config)

        self._service_tasks.append(
            asyncio.create_task(run_channel_service(self.config))
        )
        for plugin in loaded_plugins:
            for _, service_func in sorted(
                plugin.services, key=lambda s: (s[0], s[1].__name__)
            ):
                self._service_tasks.append(
                    asyncio.create_task(service_func(self._state))
                )
                logger.info(f"started service from plugin: {plugin.dist_name}")

        self._server_task = asyncio.create_task(self._server.serve())

        while not self._server.started:
            if self._server_task.done():
                await self._server_task
            await asyncio.sleep(0.1)

        return self

    async def wait(self) -> None:
        """Waits for the server task to complete (e.g. via signal or internal error)."""
        if self._server_task:
            try:
                await self._server_task
            except asyncio.CancelledError:
                pass

    async def stop(self) -> None:
        """Stops the backend runtime."""
        if self._server and self._server.started:
            self._server.should_exit = True
            if self._server_task:
                try:
                    await self._server_task
                except asyncio.CancelledError:
                    pass

        for task in self._service_tasks:
            task.cancel()
        if self._service_tasks:
            await asyncio.gather(*self._service_tasks, return_exceptions=True)
        self._service_tasks.clear()

        await self._stack.aclose()
        self._state = None
        self._server = None
        self._server_task = None

    async def __aenter__(self) -> Self:
        return await self.start()

    async def __aexit__(self, *args: object) -> None:
        await self.stop()
config instance-attribute
config = Config.from_file(config)
state property
state: BackendState
start async
start() -> Self

Starts the backend runtime.

Source code in packages/tangram_core/src/tangram_core/backend.py
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
async def start(self) -> Self:
    """Starts the backend runtime."""
    if self._state is not None:
        raise RuntimeError("runtime is already started")

    redis_client = await self._stack.enter_async_context(
        redis.from_url(self.config.core.redis_url)  # type: ignore
    )
    http_client = await self._stack.enter_async_context(
        httpx.AsyncClient(http2=True)
    )
    self._state = BackendState(
        redis_client=redis_client,
        http_client=http_client,
        config=self.config,
    )

    loaded_plugins = load_enabled_plugins(self.config)
    app = create_app(self._state, loaded_plugins)

    server_config = uvicorn.Config(
        app,
        host=self.config.server.host,
        port=self.config.server.port,
        log_config=get_log_config_dict(self.config),
    )
    self._server = uvicorn.Server(server_config)

    self._service_tasks.append(
        asyncio.create_task(run_channel_service(self.config))
    )
    for plugin in loaded_plugins:
        for _, service_func in sorted(
            plugin.services, key=lambda s: (s[0], s[1].__name__)
        ):
            self._service_tasks.append(
                asyncio.create_task(service_func(self._state))
            )
            logger.info(f"started service from plugin: {plugin.dist_name}")

    self._server_task = asyncio.create_task(self._server.serve())

    while not self._server.started:
        if self._server_task.done():
            await self._server_task
        await asyncio.sleep(0.1)

    return self
wait async
wait() -> None

Waits for the server task to complete (e.g. via signal or internal error).

Source code in packages/tangram_core/src/tangram_core/backend.py
497
498
499
500
501
502
503
async def wait(self) -> None:
    """Waits for the server task to complete (e.g. via signal or internal error)."""
    if self._server_task:
        try:
            await self._server_task
        except asyncio.CancelledError:
            pass
stop async
stop() -> None

Stops the backend runtime.

Source code in packages/tangram_core/src/tangram_core/backend.py
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
async def stop(self) -> None:
    """Stops the backend runtime."""
    if self._server and self._server.started:
        self._server.should_exit = True
        if self._server_task:
            try:
                await self._server_task
            except asyncio.CancelledError:
                pass

    for task in self._service_tasks:
        task.cancel()
    if self._service_tasks:
        await asyncio.gather(*self._service_tasks, return_exceptions=True)
    self._service_tasks.clear()

    await self._stack.aclose()
    self._state = None
    self._server = None
    self._server_task = None

get_state async

get_state(request: Request) -> BackendState
Source code in packages/tangram_core/src/tangram_core/backend.py
74
75
async def get_state(request: Request) -> BackendState:
    return request.app.state.backend_state  # type: ignore

get_distribution_path

get_distribution_path(dist_name: str) -> Path

Get the local path of a distribution, handling both editable installs (direct_url.json) and standard wheel installs.

See: https://packaging.python.org/en/latest/specifications/direct-url-data-structure/

Source code in packages/tangram_core/src/tangram_core/backend.py
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def get_distribution_path(dist_name: str) -> Path:
    """Get the local path of a distribution, handling both editable installs
    (`direct_url.json`) and standard wheel installs.

    See: https://packaging.python.org/en/latest/specifications/direct-url-data-structure/
    """
    # always try direct_url.json first (e.g. for the case of `uv sync --all-packages`)
    try:
        dist = Distribution.from_name(dist_name)
        if direct_url_content := dist.read_text("direct_url.json"):
            direct_url_data = json.loads(direct_url_content)
            if (
                (url := direct_url_data.get("url"))
                # url may point to a git or zip archive, but since we only care
                # about local paths, we only handle the file:// scheme here
                and url.startswith("file://")
            ):
                parsed = urllib.parse.urlparse(url)
                if os.name == "nt":
                    path_str = urllib.request.url2pathname(parsed.path)
                    if parsed.netloc and parsed.netloc not in ("", "localhost"):
                        path_str = f"//{parsed.netloc}{path_str}"
                    path1 = Path(path_str)
                else:
                    path1 = Path(urllib.parse.unquote(parsed.path))
                if path1.is_dir():
                    return path1
    except (PackageNotFoundError, json.JSONDecodeError, FileNotFoundError):
        pass

    # fallback in case it was installed via a wheel
    if (trav := importlib.resources.files(dist_name)).is_dir():
        with importlib.resources.as_file(trav) as path2:
            return path2
    raise FileNotFoundError(f"could not find distribution path for {dist_name}")

resolve_frontend

resolve_frontend(plugin: Plugin) -> Path | None
Source code in packages/tangram_core/src/tangram_core/backend.py
118
119
120
121
def resolve_frontend(plugin: Plugin) -> Path | None:
    if not plugin.frontend_path:
        return None
    return get_distribution_path(plugin.dist_name) / plugin.frontend_path

load_enabled_plugins

load_enabled_plugins(config: Config) -> list[Plugin]
Source code in packages/tangram_core/src/tangram_core/backend.py
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def load_enabled_plugins(
    config: Config,
) -> list[Plugin]:
    loaded_plugins = []
    enabled_plugin_names = set(config.core.plugins)

    for entry_point in scan_plugins():
        # TODO: should we check entry_point.dist.name instead?
        if entry_point.name not in enabled_plugin_names:
            continue
        if (plugin := load_plugin(entry_point)) is not None:
            loaded_plugins.append(plugin)

    return loaded_plugins

lifespan async

lifespan(app: FastAPI, backend_state: BackendState, loaded_plugins: Iterable[Plugin]) -> AsyncGenerator[None, None]
Source code in packages/tangram_core/src/tangram_core/backend.py
140
141
142
143
144
145
146
147
148
149
150
151
@asynccontextmanager
async def lifespan(
    app: FastAPI, backend_state: BackendState, loaded_plugins: Iterable[Plugin]
) -> AsyncGenerator[None, None]:
    async with AsyncExitStack() as stack:
        for plugin in loaded_plugins:
            if plugin.lifespan:
                logger.info(f"initializing lifespan for {plugin.dist_name}")
                await stack.enter_async_context(plugin.lifespan(backend_state))

        app.state.backend_state = backend_state
        yield

default_cache_dir

default_cache_dir() -> Path
Source code in packages/tangram_core/src/tangram_core/backend.py
154
155
156
157
158
159
160
161
162
def default_cache_dir() -> Path:
    if (xdg_cache := os.environ.get("XDG_CACHE_HOME")) is not None:
        cache_dir = Path(xdg_cache) / "tangram"
    else:
        cache_dir = Path(platformdirs.user_cache_dir(appname="tangram"))
    if not cache_dir.exists():
        cache_dir.mkdir(parents=True, exist_ok=True)

    return cache_dir

make_cache_route_handler

make_cache_route_handler(entry: CacheEntry, state: BackendState) -> Callable[..., Awaitable[FileResponse]]

Factory function that creates a route handler for caching and serving files. Dynamically handles URL parameters found in both serve_route and origin.

Parameters:

Name Type Description Default
entry CacheEntry

Cache entry configuration

required
state BackendState

Backend state with http_client for fetching remote resources

required

Returns:

Type Description
Callable[..., Awaitable[FileResponse]]

Async function that handles the route with dynamic parameters

Source code in packages/tangram_core/src/tangram_core/backend.py
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def make_cache_route_handler(
    entry: CacheEntry, state: BackendState
) -> Callable[..., Awaitable[FileResponse]]:
    """
    Factory function that creates a route handler for caching and serving files.
    Dynamically handles URL parameters found in both serve_route and origin.

    :param entry: Cache entry configuration
    :param state: Backend state with http_client for fetching remote resources
    :returns: Async function that handles the route with dynamic parameters
    """
    from inspect import Parameter, Signature

    # Extract parameter names from the serve_route (e.g., {fontstack}, {range})
    params = CACHE_PARAM_PATTERN.findall(entry.serve_route)

    async def cache_route_handler(**kwargs: str) -> FileResponse:
        if (local_path := entry.local_path) is None:
            local_path = default_cache_dir()
        else:
            local_path = local_path.expanduser()

        # Build the local file path by replacing parameters
        local_file = local_path
        for param in params:
            if param in kwargs:
                local_file = local_file / kwargs[param]

        logger.info(f"Serving cached file from {local_file}")

        if not local_file.exists():
            assert entry.origin is not None
            # Build the remote URL by replacing parameters
            remote_url = entry.origin
            for param, value in kwargs.items():
                remote_url = remote_url.replace(f"{{{param}}}", value)

            logger.info(f"Downloading from {remote_url} to {local_file}")
            c = await state.http_client.get(remote_url)
            c.raise_for_status()
            local_file.parent.mkdir(parents=True, exist_ok=True)
            local_file.write_bytes(c.content)

        return FileResponse(path=local_file, media_type=entry.media_type)

    # Create explicit parameters for the function signature
    sig_params = [
        Parameter(
            name=param,
            kind=Parameter.POSITIONAL_OR_KEYWORD,
            annotation=str,
        )
        for param in params
    ]
    cache_route_handler.__signature__ = Signature(  # type: ignore
        parameters=sig_params,
        return_annotation=FileResponse,
    )

    return cache_route_handler

core_into_frontend_config

core_into_frontend_config(config: Config) -> FrontendConfig
Source code in packages/tangram_core/src/tangram_core/backend.py
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
def core_into_frontend_config(config: Config) -> FrontendConfig:
    if config.channel.public_url:
        channel_url = config.channel.public_url
    else:
        host = "localhost" if config.channel.host == "0.0.0.0" else config.channel.host
        channel_url = f"http://{host}:{config.channel.port}"
    frontend_channel = FrontendChannelConfig(url=channel_url)

    user_theme_names = {t.name for t in config.core.themes}
    merged_themes = [
        t for t in DEFAULT_THEMES if t.name not in user_theme_names
    ] + config.core.themes

    frontend_core = FrontendCoreConfig(theme=config.core.theme, themes=merged_themes)
    frontend_map = replace(
        config.map,
        styles=merge_styles(default_styles(), config.map.styles),
    )

    return FrontendConfig(
        core=frontend_core, map=frontend_map, channel=frontend_channel
    )

create_app

create_app(backend_state: BackendState, loaded_plugins: Iterable[Plugin]) -> FastAPI
Source code in packages/tangram_core/src/tangram_core/backend.py
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
def create_app(
    backend_state: BackendState,
    loaded_plugins: Iterable[Plugin],
) -> FastAPI:
    app = FastAPI(
        lifespan=partial(
            lifespan, backend_state=backend_state, loaded_plugins=loaded_plugins
        ),
        default_response_class=ORJSONResponse,
    )

    frontend_config_instance = core_into_frontend_config(backend_state.config)
    frontend_adapter = TypeAdapter(FrontendConfig)
    manifest_core = to_frontend_manifest(frontend_adapter, frontend_config_instance)

    manifest_plugins = {}
    for plugin in loaded_plugins:
        backend_state.loaded_plugins[plugin.dist_name] = plugin
        for router in plugin.routers:
            app.include_router(router)

        if (frontend_path_resolved := resolve_frontend(plugin)) is None:
            continue
        app.mount(
            f"/plugins/{plugin.dist_name}",
            StaticFiles(directory=str(frontend_path_resolved)),
            name=plugin.dist_name,
        )
        if not (plugin_json_path := frontend_path_resolved / "plugin.json").exists():
            continue
        with plugin_json_path.open("rb") as f:
            plugin_meta = json.load(f)

        manifest_plugins[plugin.dist_name] = plugin_meta

        # NOTE: on the js side the `config` and `config_json_schema` keys
        # may be missing if the plugin does not configure them properly
        conf_dict = backend_state.config.plugins.get(plugin.dist_name, {})
        if (backend_adapter := plugin.adapter()) is None:
            continue
        conf_instance = backend_adapter.validate_python(conf_dict)
        if not plugin.frontend_config_class:
            continue
        if plugin.into_frontend_config_function is None:
            logger.warning(
                f"expected `into_frontend_config_function` for"
                f" plugin {plugin.dist_name} but it is None. "
                "Frontend config will be empty!"
            )
            continue
        frontend_instance = plugin.into_frontend_config_function(conf_instance)
        frontend_adapter_ = plugin.frontend_adapter()
        assert frontend_adapter_ is not None
        frontend_manifest = to_frontend_manifest(frontend_adapter_, frontend_instance)
        manifest_plugins[plugin.dist_name] = {
            **plugin_meta,
            **frontend_manifest,
        }

    @app.post("/settings/validate/{plugin_name}")
    async def validate_settings(
        plugin_name: str,
        data: dict,
        state: Annotated[BackendState, Depends(get_state)],
    ) -> dict:
        SUCCESS = {"success": True, "errors": {}}
        try:
            if plugin_name == "tangram_core":
                _ = parse_frontend_config(frontend_adapter, data)
                return SUCCESS
            plugin = state.loaded_plugins.get(plugin_name)
            assert plugin is not None, f"plugin {plugin_name} not found"

            if plugin.frontend_config_class:
                if (frontend_adapter_ := plugin.frontend_adapter()) is not None:
                    frontend_adapter_.validate_python(data)
                return SUCCESS

            # if no frontend config defined, validation always succeeds because
            # nothing to validate against
            return SUCCESS
        except ValidationError as e:
            errs = {
                ".".join(str(loc) for loc in err["loc"]): err["msg"]
                for err in e.errors()
            }
            return {"success": False, "errors": errs}

    @app.get("/manifest.json")
    async def get_manifest() -> ORJSONResponse:
        return ORJSONResponse(
            content={"core": manifest_core, "plugins": manifest_plugins}
        )

    # Cache mechanism - MUST be registered BEFORE the catch-all frontend mount
    for cache_entry in backend_state.config.cache.entries:
        logger.info(
            f"caching {cache_entry.origin} to {cache_entry.local_path} "
            f"and serving at {cache_entry.serve_route}"
        )
        route_handler = make_cache_route_handler(cache_entry, backend_state)

        logger.info(
            f"Registering route: GET {cache_entry.serve_route} with dynamic params"
        )
        app.add_api_route(
            cache_entry.serve_route,
            route_handler,
            methods=["GET"],
            name=f"cache-{cache_entry.serve_route.replace('/', '_')}",
        )

    if not (
        frontend_path := get_distribution_path("tangram_core") / "dist-frontend"
    ).is_dir():
        raise ValueError(
            f"error: frontend {frontend_path} was not found, "
            "did you run `pnpm i && pnpm run build`?"
        )
    app.mount("/", StaticFiles(directory=str(frontend_path), html=True), name="core")

    return app

run_channel_service async

run_channel_service(config: Config) -> None
Source code in packages/tangram_core/src/tangram_core/backend.py
410
411
412
413
414
415
416
417
418
419
420
421
422
423
async def run_channel_service(config: Config) -> None:
    from . import _core

    _core.init_tracing_stderr(config.core.log_level)

    rust_config = _core.ChannelConfig(
        host=config.channel.host,
        port=config.channel.port,
        redis_url=config.core.redis_url,
        jwt_secret=config.channel.jwt_secret,
        jwt_expiration_secs=config.channel.jwt_expiration_secs,
        id_length=config.channel.id_length,
    )
    await _core.run(rust_config)

get_log_config_dict

get_log_config_dict(config: Config) -> dict[str, Any]
Source code in packages/tangram_core/src/tangram_core/backend.py
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
def get_log_config_dict(config: Config) -> dict[str, Any]:
    def format_time(dt: datetime) -> str:
        return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ ")

    return {
        "version": 1,
        "disable_existing_loggers": False,
        "handlers": {
            "default": {
                "class": "rich.logging.RichHandler",
                "log_time_format": format_time,
                "omit_repeated_times": False,
            },
        },
        "root": {"handlers": ["default"], "level": config.core.log_level.upper()},
    }

config

StyleSpecification module-attribute

StyleSpecification: TypeAlias = dict[str, Any]

Url module-attribute

Url: TypeAlias = str

MapStyle module-attribute

StrOrPathLike module-attribute

StrOrPathLike = str | PathLike[str]

IntoConfig module-attribute

IntoConfig: TypeAlias = 'Config | StrOrPathLike'

HasTopbarUiConfig

Bases: Protocol

Source code in packages/tangram_core/src/tangram_core/config.py
47
48
49
@runtime_checkable
class HasTopbarUiConfig(Protocol):
    topbar_order: int
topbar_order instance-attribute
topbar_order: int

HasSidebarUiConfig

Bases: Protocol

Source code in packages/tangram_core/src/tangram_core/config.py
52
53
54
@runtime_checkable
class HasSidebarUiConfig(Protocol):
    sidebar_order: int
sidebar_order instance-attribute
sidebar_order: int

ServerConfig dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
57
58
59
60
@dataclass
class ServerConfig:
    host: str = "127.0.0.1"
    port: int = 2346
host class-attribute instance-attribute
host: str = '127.0.0.1'
port class-attribute instance-attribute
port: int = 2346

ChannelConfig dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
63
64
65
66
67
68
69
70
71
72
@dataclass
class ChannelConfig:
    # TODO: we should make it clear that host:port is for the *backend* to
    # listen on, and not to be confused with the frontend.
    host: str = "127.0.0.1"
    port: int = 2347
    public_url: str | None = None
    jwt_secret: str = "secret"
    jwt_expiration_secs: int = 315360000  # 10 years
    id_length: int = 8
host class-attribute instance-attribute
host: str = '127.0.0.1'
port class-attribute instance-attribute
port: int = 2347
public_url class-attribute instance-attribute
public_url: str | None = None
jwt_secret class-attribute instance-attribute
jwt_secret: str = 'secret'
jwt_expiration_secs class-attribute instance-attribute
jwt_expiration_secs: int = 315360000
id_length class-attribute instance-attribute
id_length: int = 8

MapConfig dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
@dataclass
class MapConfig:
    # users can specify local path in config file but it will be resolved in from_file
    # and so the stored type cannot be Path
    style: Annotated[MapStyle, FrontendMutable(widget="map-settings")] = (
        "https://basemaps.cartocdn.com/gl/voyager-gl-style/style.json"
    )
    styles: Annotated[list[MapStyle], FrontendMutable(widget="map-settings")] = field(
        default_factory=list
    )
    center_lat: Annotated[float, Ge(-90), Le(90), FrontendMutable()] = 48.0
    center_lon: Annotated[float, Ge(-180), Le(180), FrontendMutable()] = 7.0
    zoom: Annotated[float, Ge(0), Le(24), FrontendMutable()] = 4
    pitch: Annotated[float, Ge(0), Le(85), FrontendMutable()] = 0
    bearing: Annotated[float, Ge(-180), Le(180), FrontendMutable()] = 0
    lang: Annotated[str, FrontendMutable()] = "en"
    min_zoom: Annotated[float, Ge(0), Le(24), FrontendMutable()] = 0
    max_zoom: Annotated[float, Ge(0), Le(24), FrontendMutable()] = 24
    max_pitch: Annotated[float, Ge(0), Le(85), FrontendMutable()] = 70
    allow_pitch: Annotated[bool, FrontendMutable()] = True
    allow_bearing: Annotated[bool, FrontendMutable()] = True
    layers_visibility: Annotated[
        dict[str, bool] | None, FrontendMutable(widget="map-settings")
    ] = None
style class-attribute instance-attribute
style: Annotated[MapStyle, FrontendMutable(widget='map-settings')] = 'https://basemaps.cartocdn.com/gl/voyager-gl-style/style.json'
styles class-attribute instance-attribute
styles: Annotated[list[MapStyle], FrontendMutable(widget='map-settings')] = field(default_factory=list)
center_lat class-attribute instance-attribute
center_lat: Annotated[float, Ge(-90), Le(90), FrontendMutable()] = 48.0
center_lon class-attribute instance-attribute
center_lon: Annotated[float, Ge(-180), Le(180), FrontendMutable()] = 7.0
zoom class-attribute instance-attribute
zoom: Annotated[float, Ge(0), Le(24), FrontendMutable()] = 4
pitch class-attribute instance-attribute
pitch: Annotated[float, Ge(0), Le(85), FrontendMutable()] = 0
bearing class-attribute instance-attribute
bearing: Annotated[float, Ge(-180), Le(180), FrontendMutable()] = 0
lang class-attribute instance-attribute
lang: Annotated[str, FrontendMutable()] = 'en'
min_zoom class-attribute instance-attribute
min_zoom: Annotated[float, Ge(0), Le(24), FrontendMutable()] = 0
max_zoom class-attribute instance-attribute
max_zoom: Annotated[float, Ge(0), Le(24), FrontendMutable()] = 24
max_pitch class-attribute instance-attribute
max_pitch: Annotated[float, Ge(0), Le(85), FrontendMutable()] = 70
allow_pitch class-attribute instance-attribute
allow_pitch: Annotated[bool, FrontendMutable()] = True
allow_bearing class-attribute instance-attribute
allow_bearing: Annotated[bool, FrontendMutable()] = True
layers_visibility class-attribute instance-attribute
layers_visibility: Annotated[dict[str, bool] | None, FrontendMutable(widget='map-settings')] = None

ThemeDefinition dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
165
166
167
168
169
170
171
172
173
174
175
176
177
178
@dataclass
class ThemeDefinition:
    name: str
    background: Annotated[str, FrontendMutable(kind="color")]
    foreground: Annotated[str, FrontendMutable(kind="color")]
    surface: Annotated[str, FrontendMutable(kind="color")]
    border: Annotated[str, FrontendMutable(kind="color")]
    hover: Annotated[str, FrontendMutable(kind="color")]
    accent1: Annotated[str, FrontendMutable(kind="color")]
    accent1_foreground: Annotated[str, FrontendMutable(kind="color")]
    accent2: Annotated[str, FrontendMutable(kind="color")]
    accent2_foreground: Annotated[str, FrontendMutable(kind="color")]
    muted: Annotated[str, FrontendMutable(kind="color")]
    error: Annotated[str, FrontendMutable(kind="color")]
name instance-attribute
name: str
background instance-attribute
background: Annotated[str, FrontendMutable(kind='color')]
foreground instance-attribute
foreground: Annotated[str, FrontendMutable(kind='color')]
surface instance-attribute
surface: Annotated[str, FrontendMutable(kind='color')]
border instance-attribute
border: Annotated[str, FrontendMutable(kind='color')]
hover instance-attribute
hover: Annotated[str, FrontendMutable(kind='color')]
accent1 instance-attribute
accent1: Annotated[str, FrontendMutable(kind='color')]
accent1_foreground instance-attribute
accent1_foreground: Annotated[str, FrontendMutable(kind='color')]
accent2 instance-attribute
accent2: Annotated[str, FrontendMutable(kind='color')]
accent2_foreground instance-attribute
accent2_foreground: Annotated[str, FrontendMutable(kind='color')]
muted instance-attribute
muted: Annotated[str, FrontendMutable(kind='color')]
error instance-attribute
error: Annotated[str, FrontendMutable(kind='color')]

AdaptiveTheme dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
181
182
183
184
@dataclass
class AdaptiveTheme:
    light: Annotated[str, FrontendMutable()] = "light"
    dark: Annotated[str, FrontendMutable()] = "dark"
light class-attribute instance-attribute
light: Annotated[str, FrontendMutable()] = 'light'
dark class-attribute instance-attribute
dark: Annotated[str, FrontendMutable()] = 'dark'

CoreConfig dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
187
188
189
190
191
192
193
194
195
196
197
@dataclass
class CoreConfig:
    redis_url: str = "redis://127.0.0.1:6379"
    plugins: list[str] = field(default_factory=list)
    log_level: str = "INFO"
    theme: Annotated[str | AdaptiveTheme, FrontendMutable(widget="theme-settings")] = (
        field(default_factory=AdaptiveTheme)
    )
    themes: Annotated[
        list[ThemeDefinition], FrontendMutable(widget="theme-settings")
    ] = field(default_factory=list)
redis_url class-attribute instance-attribute
redis_url: str = 'redis://127.0.0.1:6379'
plugins class-attribute instance-attribute
plugins: list[str] = field(default_factory=list)
log_level class-attribute instance-attribute
log_level: str = 'INFO'
theme class-attribute instance-attribute
theme: Annotated[str | AdaptiveTheme, FrontendMutable(widget='theme-settings')] = field(default_factory=AdaptiveTheme)
themes class-attribute instance-attribute
themes: Annotated[list[ThemeDefinition], FrontendMutable(widget='theme-settings')] = field(default_factory=list)

CacheEntry dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
200
201
202
203
204
205
206
207
208
209
@dataclass
class CacheEntry:
    origin: str | None = None
    """Origin URL. If None, the local file is served directly."""
    local_path: Path | None = None
    """Local path to cache the file."""
    serve_route: str = ""
    """Where to serve the file in FastAPI."""
    media_type: str = "application/octet-stream"
    """Media type for the response."""
origin class-attribute instance-attribute
origin: str | None = None

Origin URL. If None, the local file is served directly.

local_path class-attribute instance-attribute
local_path: Path | None = None

Local path to cache the file.

serve_route class-attribute instance-attribute
serve_route: str = ''

Where to serve the file in FastAPI.

media_type class-attribute instance-attribute
media_type: str = 'application/octet-stream'

Media type for the response.

CacheConfig dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
212
213
214
@dataclass
class CacheConfig:
    entries: list[CacheEntry] = field(default_factory=list)
entries class-attribute instance-attribute
entries: list[CacheEntry] = field(default_factory=list)

Config dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
@dataclass
class Config:
    core: CoreConfig = field(default_factory=CoreConfig)
    server: ServerConfig = field(default_factory=ServerConfig)
    channel: ChannelConfig = field(default_factory=ChannelConfig)
    map: MapConfig = field(default_factory=MapConfig)
    plugins: dict[str, Any] = field(default_factory=dict)
    """Mapping of plugin name to plugin-specific config."""
    cache: CacheConfig = field(default_factory=CacheConfig)

    @classmethod
    def from_file(cls, config_path: StrOrPathLike) -> Config:
        if sys.version_info < (3, 11):
            import tomli as tomllib
        else:
            import tomllib
        from pydantic import TypeAdapter

        with open(config_path, "rb") as f:
            cfg_data = tomllib.load(f)

        base_dir = Path(config_path).parent
        map = cfg_data.setdefault("map", {})
        if (s := map.get("style", None)) is not None:
            map["style"] = try_resolve_local_style(base_dir, s)

        if "styles" in map:
            map["styles"] = [
                try_resolve_local_style(base_dir, style)
                for style in map.get("styles", [])
            ]

        config_adapter = TypeAdapter(cls)
        config = config_adapter.validate_python(cfg_data)
        return config
core class-attribute instance-attribute
core: CoreConfig = field(default_factory=CoreConfig)
server class-attribute instance-attribute
server: ServerConfig = field(default_factory=ServerConfig)
channel class-attribute instance-attribute
channel: ChannelConfig = field(default_factory=ChannelConfig)
map class-attribute instance-attribute
map: MapConfig = field(default_factory=MapConfig)
plugins class-attribute instance-attribute
plugins: dict[str, Any] = field(default_factory=dict)

Mapping of plugin name to plugin-specific config.

cache class-attribute instance-attribute
cache: CacheConfig = field(default_factory=CacheConfig)
from_file classmethod
from_file(config_path: StrOrPathLike) -> Config
Source code in packages/tangram_core/src/tangram_core/config.py
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
@classmethod
def from_file(cls, config_path: StrOrPathLike) -> Config:
    if sys.version_info < (3, 11):
        import tomli as tomllib
    else:
        import tomllib
    from pydantic import TypeAdapter

    with open(config_path, "rb") as f:
        cfg_data = tomllib.load(f)

    base_dir = Path(config_path).parent
    map = cfg_data.setdefault("map", {})
    if (s := map.get("style", None)) is not None:
        map["style"] = try_resolve_local_style(base_dir, s)

    if "styles" in map:
        map["styles"] = [
            try_resolve_local_style(base_dir, style)
            for style in map.get("styles", [])
        ]

    config_adapter = TypeAdapter(cls)
    config = config_adapter.validate_python(cfg_data)
    return config

FrontendCoreConfig dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
279
280
281
282
283
284
285
286
@dataclass
class FrontendCoreConfig:
    theme: Annotated[str | AdaptiveTheme, FrontendMutable(widget="theme-settings")] = (
        field(default_factory=AdaptiveTheme)
    )
    themes: Annotated[
        list[ThemeDefinition], FrontendMutable(widget="theme-settings")
    ] = field(default_factory=list)
theme class-attribute instance-attribute
theme: Annotated[str | AdaptiveTheme, FrontendMutable(widget='theme-settings')] = field(default_factory=AdaptiveTheme)
themes class-attribute instance-attribute
themes: Annotated[list[ThemeDefinition], FrontendMutable(widget='theme-settings')] = field(default_factory=list)

FrontendChannelConfig dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
289
290
291
@dataclass
class FrontendChannelConfig:
    url: str
url instance-attribute
url: str

FrontendConfig dataclass

Source code in packages/tangram_core/src/tangram_core/config.py
294
295
296
297
298
@dataclass
class FrontendConfig:
    core: FrontendCoreConfig
    map: MapConfig
    channel: FrontendChannelConfig
core instance-attribute
map instance-attribute
map: MapConfig
channel instance-attribute

FrontendMutable dataclass

Marker to allow a particular field in a frontend configuration to be mutated in the settings page.

Source code in packages/tangram_core/src/tangram_core/config.py
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
@dataclass(frozen=True)
class FrontendMutable:
    """Marker to allow a particular field in a frontend configuration to be mutated
    in the settings page."""

    # TODO maybe in the future we can merge the two
    kind: Literal["color"] | None = None
    widget: str | None = None
    """Identifier for a custom widget to use for this field in the frontend UI.
    If multiple fields share the same widget id, they will be grouped together.
    """

    def __get_pydantic_json_schema__(
        self, core_schema: CoreSchema, handler: GetJsonSchemaHandler
    ) -> JsonSchemaValue:
        json_schema = handler(core_schema)
        json_schema["tangram_mutable"] = True
        if self.kind:
            json_schema["tangram_kind"] = self.kind
        if self.widget:
            json_schema["tangram_widget"] = self.widget
        return json_schema
kind class-attribute instance-attribute
kind: Literal['color'] | None = None
widget class-attribute instance-attribute
widget: str | None = None

Identifier for a custom widget to use for this field in the frontend UI. If multiple fields share the same widget id, they will be grouped together.

FrontendData

Bases: TypedDict

Source code in packages/tangram_core/src/tangram_core/config.py
330
331
332
class FrontendData(TypedDict):
    config: dict[str, Any]
    config_json_schema: dict[str, Any]
config instance-attribute
config: dict[str, Any]
config_json_schema instance-attribute
config_json_schema: dict[str, Any]

default_config_file

default_config_file() -> Path
Source code in packages/tangram_core/src/tangram_core/config.py
34
35
36
37
38
39
40
41
42
43
44
def default_config_file() -> Path:
    import platformdirs

    if (xdg_config := os.environ.get("XDG_CONFIG_HOME")) is not None:
        config_dir = Path(xdg_config) / "tangram"
    else:
        config_dir = Path(platformdirs.user_config_dir(appname="tangram"))
    if not config_dir.exists():
        config_dir.mkdir(parents=True, exist_ok=True)

    return Path(config_dir) / "tangram.toml"

get_default_asset

get_default_asset(*parts: str) -> Traversable
Source code in packages/tangram_core/src/tangram_core/config.py
81
82
83
84
85
def get_default_asset(*parts: str) -> Traversable:
    asset = importlib.resources.files("tangram_core").joinpath("assets")
    for part in parts:
        asset = asset.joinpath(part)
    return asset

load_built_in_style

load_built_in_style(filename: str) -> StyleSpecification
Source code in packages/tangram_core/src/tangram_core/config.py
88
89
90
91
92
def load_built_in_style(filename: str) -> StyleSpecification:
    style: StyleSpecification = json.loads(
        get_default_asset("map_styles", filename).read_text(encoding="utf-8")
    )
    return style

default_styles

default_styles() -> list[MapStyle]
Source code in packages/tangram_core/src/tangram_core/config.py
 95
 96
 97
 98
 99
100
101
def default_styles() -> list[MapStyle]:
    return [
        "https://basemaps.cartocdn.com/gl/voyager-gl-style/style.json",
        "https://basemaps.cartocdn.com/gl/positron-gl-style/style.json",
        "https://basemaps.cartocdn.com/gl/dark-matter-gl-style/style.json",
        load_built_in_style("radar-dark.json"),
    ]

style_identifier

style_identifier(style: StyleSpecification, key: str) -> str | None
Source code in packages/tangram_core/src/tangram_core/config.py
104
105
106
def style_identifier(style: StyleSpecification, key: str) -> str | None:
    value = style.get(key)
    return value if isinstance(value, str) else None

styles_match

styles_match(candidate: MapStyle, override: MapStyle) -> bool
Source code in packages/tangram_core/src/tangram_core/config.py
109
110
111
112
113
114
115
116
117
118
def styles_match(candidate: MapStyle, override: MapStyle) -> bool:
    if isinstance(override, str):
        return candidate == override
    if not isinstance(candidate, dict):
        return False
    if (identifier := style_identifier(override, "id")) is not None:
        return style_identifier(candidate, "id") == identifier
    if (name := style_identifier(override, "name")) is not None:
        return style_identifier(candidate, "name") == name
    return False

merge_styles

merge_styles(defaults: list[MapStyle], overrides: list[MapStyle]) -> list[MapStyle]
Source code in packages/tangram_core/src/tangram_core/config.py
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def merge_styles(defaults: list[MapStyle], overrides: list[MapStyle]) -> list[MapStyle]:
    merged = list(defaults)
    for override in overrides:
        position = next(
            (
                index
                for index, candidate in enumerate(merged)
                if styles_match(candidate, override)
            ),
            None,
        )
        if position is None:
            merged.append(override)
        else:
            merged[position] = override
    return merged

try_resolve_local_style

try_resolve_local_style(base_dir: Path, style: MapStyle) -> MapStyle
Source code in packages/tangram_core/src/tangram_core/config.py
260
261
262
263
264
265
266
267
268
269
def try_resolve_local_style(base_dir: Path, style: MapStyle) -> MapStyle:
    if isinstance(style, str):
        scheme = urllib.parse.urlparse(style).scheme
        if scheme in ("http", "https"):
            return style
        if (p := (base_dir / style).resolve()).is_file():
            with open(p, "rb") as f:
                specification: StyleSpecification = json.load(f)
            return specification
    return style

to_frontend_manifest

to_frontend_manifest(adapter: TypeAdapter, frontend_cfg: Any) -> FrontendData

Serialises a frontend configuration.

Source code in packages/tangram_core/src/tangram_core/config.py
335
336
337
338
339
def to_frontend_manifest(adapter: TypeAdapter, frontend_cfg: Any) -> FrontendData:
    """Serialises a frontend configuration."""
    config_dump = adapter.dump_python(frontend_cfg)
    config_json_schema = adapter.json_schema()
    return {"config": config_dump, "config_json_schema": config_json_schema}

parse_frontend_config

parse_frontend_config(adapter: TypeAdapter, frontend_cfg: Any) -> Any

Deserialises and validates frontend-submitted config data.

Source code in packages/tangram_core/src/tangram_core/config.py
342
343
344
def parse_frontend_config(adapter: TypeAdapter, frontend_cfg: Any) -> Any:
    """Deserialises and validates frontend-submitted config data."""
    return adapter.validate_python(frontend_cfg)

plugin

ServiceAsyncFunc module-attribute

ServiceAsyncFunc: TypeAlias = Callable[[BackendState], Coroutine[Any, Any, None]]

ServiceFunc module-attribute

ServiceFunc: TypeAlias = ServiceAsyncFunc | Callable[[BackendState], None]

Priority module-attribute

Priority: TypeAlias = int

IntoFrontendConfigFunction module-attribute

IntoFrontendConfigFunction: TypeAlias = Callable[[Any], Any]

Lifespan module-attribute

logger module-attribute

logger = logging.getLogger(__name__)

Plugin dataclass

Stores the metadata and registered API routes, background services and frontend assets for a tangram plugin.

Packages should declare an entry point in the tangram_core.plugins group in their pyproject.toml pointing to an instance of this class.

Source code in packages/tangram_core/src/tangram_core/plugin.py
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@dataclass(eq=False)
class Plugin:
    """Stores the metadata and registered API routes, background services and
    frontend assets for a tangram plugin.

    Packages should declare an entry point in the `tangram_core.plugins` group
    in their `pyproject.toml` pointing to an instance of this class.
    """

    frontend_path: str | None = None
    """Path to the compiled frontend assets, *relative* to the distribution root
    (editable) or package root (wheel).
    """
    routers: list[APIRouter] = field(default_factory=list)
    config_class: type | None = None
    """The **backend** configuration class (dataclass, TypedDict or Pydantic model)
    for the plugin."""
    frontend_config_class: type | None = None
    """The **frontend** configuration class for the plugin. If set, it will be used to
    generate the frontend schema and validate settings updates.
    Fields should be annotated with [tangram_core.config.FrontendMutable][] if they
    are allowed to be modified from the frontend settings UI.
    """
    into_frontend_config_function: IntoFrontendConfigFunction | None = None
    """Function to transform the backend configuration into the frontend
    configuration. It receives the validated backend configuration object and
    should return an instance of `frontend_config_class`.
    Useful if the user wants to hide sensitive fields (e.g. API keys) from the frontend
    or dynamically compute certain fields.
    Required if `frontend_config_class` is set.
    """
    lifespan: Lifespan | None = None
    """Async context manager for plugin initialization and teardown."""
    services: list[tuple[Priority, ServiceAsyncFunc]] = field(
        default_factory=list, init=False
    )
    get_typer: Callable[[], Typer] | None = None
    """A function that returns the subcommands which will later be registered."""
    dist_name: str = field(init=False)
    """Name of the distribution (package) that provided this plugin, populated
    automatically during loading.
    """  # we do this so plugins can know their own package name if needed

    def register_service(
        self, priority: Priority = 0
    ) -> Callable[[ServiceFunc], ServiceFunc]:
        """Decorator to register a background service function.

        Services are long-running async functions that receive the BackendState
        and are started when the application launches.
        """

        def decorator(func: ServiceFunc) -> ServiceFunc:
            @functools.wraps(func)
            async def async_wrapper(backend_state: BackendState) -> None:
                if asyncio.iscoroutinefunction(func):
                    await func(backend_state)
                else:
                    await asyncio.to_thread(func, backend_state)

            self.services.append((priority, async_wrapper))
            return func

        return decorator

    # keeping the public methods callable for plugin API compatibility.
    # plugins are mutable registries, so equality and hashing use object identity
    # method-level lru_cache would hash and retain mutable plugins

    @functools.cached_property
    def _adapter(self) -> TypeAdapter[Any] | None:
        from pydantic import TypeAdapter

        if self.config_class:
            return TypeAdapter(self.config_class)
        return None

    def adapter(self) -> TypeAdapter[Any] | None:
        """Returns a cached Pydantic TypeAdapter for the plugin's configuration class.

        Avoids expensive rebuilds on every validation request, such as those from the
        settings UI.
        """
        return self._adapter

    @functools.cached_property
    def _frontend_adapter(self) -> TypeAdapter[Any] | None:
        from pydantic import TypeAdapter

        if self.frontend_config_class:
            return TypeAdapter(self.frontend_config_class)
        return None

    def frontend_adapter(self) -> TypeAdapter[Any] | None:
        """Returns a cached Pydantic TypeAdapter for the plugin's frontend
        configuration class."""
        return self._frontend_adapter
frontend_path class-attribute instance-attribute
frontend_path: str | None = None

Path to the compiled frontend assets, relative to the distribution root (editable) or package root (wheel).

routers class-attribute instance-attribute
routers: list[APIRouter] = field(default_factory=list)
config_class class-attribute instance-attribute
config_class: type | None = None

The backend configuration class (dataclass, TypedDict or Pydantic model) for the plugin.

frontend_config_class class-attribute instance-attribute
frontend_config_class: type | None = None

The frontend configuration class for the plugin. If set, it will be used to generate the frontend schema and validate settings updates. Fields should be annotated with tangram_core.config.FrontendMutable if they are allowed to be modified from the frontend settings UI.

into_frontend_config_function class-attribute instance-attribute
into_frontend_config_function: IntoFrontendConfigFunction | None = None

Function to transform the backend configuration into the frontend configuration. It receives the validated backend configuration object and should return an instance of frontend_config_class. Useful if the user wants to hide sensitive fields (e.g. API keys) from the frontend or dynamically compute certain fields. Required if frontend_config_class is set.

lifespan class-attribute instance-attribute
lifespan: Lifespan | None = None

Async context manager for plugin initialization and teardown.

services class-attribute instance-attribute
services: list[tuple[Priority, ServiceAsyncFunc]] = field(default_factory=list, init=False)
get_typer class-attribute instance-attribute
get_typer: Callable[[], Typer] | None = None

A function that returns the subcommands which will later be registered.

dist_name class-attribute instance-attribute
dist_name: str = field(init=False)

Name of the distribution (package) that provided this plugin, populated automatically during loading.

register_service
register_service(priority: Priority = 0) -> Callable[[ServiceFunc], ServiceFunc]

Decorator to register a background service function.

Services are long-running async functions that receive the BackendState and are started when the application launches.

Source code in packages/tangram_core/src/tangram_core/plugin.py
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def register_service(
    self, priority: Priority = 0
) -> Callable[[ServiceFunc], ServiceFunc]:
    """Decorator to register a background service function.

    Services are long-running async functions that receive the BackendState
    and are started when the application launches.
    """

    def decorator(func: ServiceFunc) -> ServiceFunc:
        @functools.wraps(func)
        async def async_wrapper(backend_state: BackendState) -> None:
            if asyncio.iscoroutinefunction(func):
                await func(backend_state)
            else:
                await asyncio.to_thread(func, backend_state)

        self.services.append((priority, async_wrapper))
        return func

    return decorator
adapter
adapter() -> TypeAdapter[Any] | None

Returns a cached Pydantic TypeAdapter for the plugin's configuration class.

Avoids expensive rebuilds on every validation request, such as those from the settings UI.

Source code in packages/tangram_core/src/tangram_core/plugin.py
109
110
111
112
113
114
115
def adapter(self) -> TypeAdapter[Any] | None:
    """Returns a cached Pydantic TypeAdapter for the plugin's configuration class.

    Avoids expensive rebuilds on every validation request, such as those from the
    settings UI.
    """
    return self._adapter
frontend_adapter
frontend_adapter() -> TypeAdapter[Any] | None

Returns a cached Pydantic TypeAdapter for the plugin's frontend configuration class.

Source code in packages/tangram_core/src/tangram_core/plugin.py
125
126
127
128
def frontend_adapter(self) -> TypeAdapter[Any] | None:
    """Returns a cached Pydantic TypeAdapter for the plugin's frontend
    configuration class."""
    return self._frontend_adapter

scan_plugins

scan_plugins() -> EntryPoints
Source code in packages/tangram_core/src/tangram_core/plugin.py
131
132
def scan_plugins() -> importlib.metadata.EntryPoints:
    return importlib.metadata.entry_points(group="tangram_core.plugins")

load_plugin

load_plugin(entry_point: EntryPoint) -> Plugin | None

Instantiates the plugin object defined in the entry point and injects the name of the distribution into it.

Source code in packages/tangram_core/src/tangram_core/plugin.py
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 load_plugin(
    entry_point: importlib.metadata.EntryPoint,
) -> Plugin | None:
    """Instantiates the plugin object defined in the entry point
    and injects the name of the distribution into it."""
    try:
        plugin_instance = entry_point.load()
    except Exception as e:  # ruff: ignore[BLE001] plugin imports may raise arbitrary exceptions
        tb = traceback.format_exc()
        logger.error(
            f"failed to load plugin {entry_point.name}: {e}. {tb}"
            f"\n= help: does {entry_point.value} exist?"
        )
        return None
    if not isinstance(plugin_instance, Plugin):
        logger.error(f"entry point {entry_point.name} is not an instance of `Plugin`")
        return None
    if entry_point.dist is None:
        logger.error(f"could not determine distribution for plugin {entry_point.name}")
        return None
    # NOTE: we ignore `entry_point.name` for now and simply use the distribution's name
    # should we raise an error if they differ? not for now

    plugin_instance.dist_name = entry_point.dist.name
    return plugin_instance

redis

log module-attribute

log = logging.getLogger(__name__)

StateT module-attribute

StateT = TypeVar('StateT')

Subscriber

Bases: ABC, Generic[StateT]

Source code in packages/tangram_core/src/tangram_core/redis.py
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
class Subscriber(abc.ABC, Generic[StateT]):
    redis: Redis
    task: asyncio.Task[None]
    pubsub: PubSub

    def __init__(
        self, name: str, redis_url: str, channels: list[str], initial_state: StateT
    ):
        self.name = name
        self.redis_url: str = redis_url
        self.channels: list[str] = channels
        self.state: StateT = initial_state
        self._running = False

    async def subscribe(self) -> None:
        if self._running:
            log.warning("%s already running", self.name)
            return

        try:
            self.redis = await Redis.from_url(self.redis_url)
            self.pubsub = self.redis.pubsub()
            await self.pubsub.psubscribe(*self.channels)
        except RedisError as e:
            log.error("%s failed to connect to Redis: %s", self.name, e)
            raise

        async def listen() -> None:
            try:
                log.info("%s listening ...", self.name)
                async for message in self.pubsub.listen():
                    log.debug("message: %s", message)
                    if message["type"] == "pmessage":
                        await self.message_handler(
                            message["channel"].decode("utf-8"),
                            message["data"].decode("utf-8"),
                            message["pattern"].decode("utf-8"),
                            self.state,
                        )
            except asyncio.CancelledError:
                log.warning("%s cancelled", self.name)

        self._running = True

        self.task = asyncio.create_task(listen())
        log.info("%s task created, running ...", self.name)

    async def cleanup(self) -> None:
        if not self._running:
            return

        if self.task:
            log.debug("%s canceling task ...", self.name)
            self.task.cancel()
            try:
                log.debug("%s await task to finish ...", self.name)
                await self.task
                log.debug("%s task canceled", self.name)
            except asyncio.CancelledError as exc:
                log.error("%s task canceling error: %s", self.name, exc)
        if self.pubsub:
            await self.pubsub.unsubscribe()
        if self.redis:
            await self.redis.close()
        self._running = False

    def is_active(self) -> bool:
        """Return True if the subscriber is actively listening."""
        return self._running and self.task is not None and not self.task.done()

    @abc.abstractmethod
    async def message_handler(
        self, event: str, payload: str, pattern: str, state: StateT
    ) -> None:
        pass
redis instance-attribute
redis: Redis
task instance-attribute
task: Task[None]
pubsub instance-attribute
pubsub: PubSub
name instance-attribute
name = name
redis_url instance-attribute
redis_url: str = redis_url
channels instance-attribute
channels: list[str] = channels
state instance-attribute
state: StateT = initial_state
subscribe async
subscribe() -> None
Source code in packages/tangram_core/src/tangram_core/redis.py
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
async def subscribe(self) -> None:
    if self._running:
        log.warning("%s already running", self.name)
        return

    try:
        self.redis = await Redis.from_url(self.redis_url)
        self.pubsub = self.redis.pubsub()
        await self.pubsub.psubscribe(*self.channels)
    except RedisError as e:
        log.error("%s failed to connect to Redis: %s", self.name, e)
        raise

    async def listen() -> None:
        try:
            log.info("%s listening ...", self.name)
            async for message in self.pubsub.listen():
                log.debug("message: %s", message)
                if message["type"] == "pmessage":
                    await self.message_handler(
                        message["channel"].decode("utf-8"),
                        message["data"].decode("utf-8"),
                        message["pattern"].decode("utf-8"),
                        self.state,
                    )
        except asyncio.CancelledError:
            log.warning("%s cancelled", self.name)

    self._running = True

    self.task = asyncio.create_task(listen())
    log.info("%s task created, running ...", self.name)
cleanup async
cleanup() -> None
Source code in packages/tangram_core/src/tangram_core/redis.py
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
async def cleanup(self) -> None:
    if not self._running:
        return

    if self.task:
        log.debug("%s canceling task ...", self.name)
        self.task.cancel()
        try:
            log.debug("%s await task to finish ...", self.name)
            await self.task
            log.debug("%s task canceled", self.name)
        except asyncio.CancelledError as exc:
            log.error("%s task canceling error: %s", self.name, exc)
    if self.pubsub:
        await self.pubsub.unsubscribe()
    if self.redis:
        await self.redis.close()
    self._running = False
is_active
is_active() -> bool

Return True if the subscriber is actively listening.

Source code in packages/tangram_core/src/tangram_core/redis.py
81
82
83
def is_active(self) -> bool:
    """Return True if the subscriber is actively listening."""
    return self._running and self.task is not None and not self.task.done()
message_handler abstractmethod async
message_handler(event: str, payload: str, pattern: str, state: StateT) -> None
Source code in packages/tangram_core/src/tangram_core/redis.py
85
86
87
88
89
@abc.abstractmethod
async def message_handler(
    self, event: str, payload: str, pattern: str, state: StateT
) -> None:
    pass

tangram_core._core

ChannelConfig

host property writable

host: str

id_length property writable

id_length: int

jwt_expiration_secs property writable

jwt_expiration_secs: int

jwt_secret property writable

jwt_secret: str

port property writable

port: int

redis_url property writable

redis_url: str

__new__

__new__(host: str, port: int, redis_url: str, jwt_secret: str, jwt_expiration_secs: int, id_length: int) -> ChannelConfig

init_tracing_stderr

init_tracing_stderr(filter_str: str) -> None

run

run(config: ChannelConfig) -> Any