atoti_limits

Plugin to connect an Atoti session to a running Limits server. Atoti server and the Limits server authenticate their requests to each other with JWTs signed by a shared key pair, so both sides must be configured with the same keys:

>>> key_pair = _generate_key_pair()
>>> jwt_config = tt.JwtConfig(key_pair=key_pair)

Starting the Limits server:

>>> atoti_port = 1337

>>> # `_start_limits_server()` is a test helper, not part of the public API.
>>> # A real project would deploy and run the standalone Limits server itself.
>>> limits_server = _start_limits_server(
...     # Set the `activeviam.apps.inter-server-event-service.issuer.target-servers` URL property.
...     atoti_url=f"http://localhost:{atoti_port}",
...     # Set the `atoti.jwt.key.public` and `atoti.jwt.key.private` properties.
...     jwt_config=jwt_config,
... )

>>> from atoti_limits import LimitsConfig
>>> session_config = tt.SessionConfig(
...     limits=LimitsConfig(url=limits_server.url),
...     port=atoti_port,
...     security=tt.SecurityConfig(jwt=jwt_config),
... )
>>> session = tt.Session.start(session_config)

Configuring the data model, giving the cube a slicing date hierarchy since Atoti Limits evaluates limits at specific points in time:

>>> sales_df = pd.DataFrame(
...     {
...         "ID": ["S0", "S1"],
...         "Date": pd.to_datetime(["2026-01-30", "2026-01-31"]),
...         "Quantity": [1.0, 2.0],
...     }
... )
>>> table = session.read_pandas(sales_df, keys={"ID"}, table_name="Example")
>>> cube = session.create_cube(table)
>>> cube.hierarchies["Date"].slicing = True

Configuring the technical user expected by the Limits server:

>>> admin_username, admin_password = "admin", "admin"
>>> session.security.individual_roles[admin_username] = {
...     "ROLE_ACTIVITI_ADMIN",
...     "ROLE_ACTIVITI_USER",
...     "ROLE_ADMIN",
...     "ROLE_AUTO_APPROVE_LIMIT",
...     "ROLE_LIMITS",
...     "ROLE_MANAGERS",
...     "ROLE_USER",
...     "ROLE_USERS",
... }
>>> session.security.basic_authentication.credentials[admin_username] = admin_password

Initiating the connection:

>>> import httpx2
>>> _ = httpx2.post(
...     f"{session.url}/limits/autoconfig/connect",
...     auth=(admin_username, admin_password),
...     json={
...         "atotiBaseUrl": session.url,
...         "limitsBaseUrl": limits_server.url,
...         # Must match the name the Atoti Limits server knows this session by.
...         "serverName": "atoti",
...     },
... ).raise_for_status()

The Limits server connects back asynchronously. Polling until the connection is established:

>>> @_poll()
... def check_connected(expect_connected: bool, /) -> bool:
...     response = httpx2.get(
...         f"{session.url}/limits/autoconfig/connected",
...         auth=(admin_username, admin_password),
...     ).raise_for_status()
...     connected = response.json()
...     assert isinstance(connected, bool), "Unexpected response body."
...     assert connected == expect_connected
...     return connected
>>> check_connected(True)
True

Once connected, the Limits pages in Atoti UI served from atoti.Session.url can be used. The Atoti session can also be disconnected from the Limits server:

>>> _ = httpx2.post(
...     f"{session.url}/limits/autoconfig/disconnect",
...     auth=(admin_username, admin_password),
... ).raise_for_status()
>>> check_connected(False)
False


`LimitsConfig`	Configuration for connecting to an Atoti Limits server.

Releases and upgrades

Getting started

Guides

API reference