Topics¶

diffusion.features.topics.Topics ¶

Bases: Component

Topics component.

It is not supposed to be instantiated independently; an instance is available on each Session instance as session.topics.

topics `property` ¶

topics: dict

Internal storage for registered topics.

fetch_request ¶

fetch_request() -> FetchRequest[IVoidFetch]

Gets an unconfigured fetch request.

If the request is invoked by calling FetchRequest.fetch, the fetch result will provide the paths and types of all of the topics which the session has permission to read.

You will usually want to restrict the query to a subset of the topic tree, and to retrieve the topic values and/or properties. This is achieved by applying one or more of the fluent builder methods to produce more refined requests.

For example:

>>> result = asyncio.run(
>>>    diffusion.sessions().open(DIFFUSION_URL).session.topics.
>>>    fetch_request().with_values().fetch( "A/B//" )
>>> )

Returns: The unconfigured fetch request. See Also: FetchRequest

add_topic `async` ¶

add_topic(
    topic_path: str, specification: DataTypeArgument
) -> TopicAddResponse

Create a new topic of the given type and properties.

Parameters:

Name	Type	Description	Default
`topic_path`	`str`	The path to create the topic on.	required
`specification`	`DataTypeArgument`	Data type of the topic.	required

add_and_set_topic `async` ¶

add_and_set_topic(
    topic_path: str,
    specification: DataTypeArgument,
    value: Any,
    constraint: typing.Optional[UpdateConstraint] = None,
) -> TopicAddResponse

Create a new topic of the given type and properties.

Parameters:

Name	Type	Description	Default
`topic_path`	`str`	The path to create the topic on.	required
`specification`	`DataTypeArgument`	Data type of the topic.	required
`value`	`Any`	Value to set when creating the topic. The value needs to be compatible with the `topic_type`. If the topic already exists, this will be set as its value.	required
`constraint`	`typing.Optional[UpdateConstraint]`	optional update constraint	`None`

set_topic `async` ¶

set_topic(
    topic_path: str,
    value: Any,
    specification: DataTypeArgument,
    constraint: typing.Optional[UpdateConstraint] = None,
) -> None

Sets the topic to a specified value.

None can only be passed to the value parameter when updating STRING, INT64, or DOUBLE topics.

When a STRING, INT64, or DOUBLE topic is set to None, the topic will be updated to have no value. If a previous value was present subscribers will receive a notification that the new value is None. New subscribers will not receive a value notification.

Parameters:

Name	Type	Description	Default
`topic_path`	`str`	the path of the topic	required
`specification`	`DataTypeArgument`	Data type of the topic.	required
`value`	`Any`	the value. `STRING`, `INT64`, or `DOUBLE` topics accept `None`, as described above. Using `None` with other topic types is an error and will throw an `UnknownDataTypeError`.	required
`constraint`	`typing.Optional[UpdateConstraint]`	optional update constraint	`None`

remove_topic `async` ¶

remove_topic(topic_selector: str) -> TopicRemovalResult

Remove all the topics that match the given selector.

Parameters:

Name	Type	Description	Default
`topic_selector`	`str`	The topics matching this selector will be removed from the server.	required

Returns:

Type	Description
`TopicRemovalResult`	An object detailing the results.

add_value_stream ¶

add_value_stream(
    topic_selector: str, stream: ValueStreamHandler
) -> None

Registers a value stream handler for a topic selector.

A value stream is a series of events associated with a registered topic. This method adds a ValueStreamHandler which can handle those events.

Parameters:

Name	Type	Description	Default
`topic_selector`	`str`	The handler will react to the updates to all topics matching this selector.	required
`stream`	`ValueStreamHandler`	A handler for incoming events of the matching data type.	required

add_fallback_stream ¶

add_fallback_stream(stream: ValueStreamHandler) -> None

Registers a fallback stream handler for a topic type.

A value stream is a series of events associated with a registered topic. This method makes it possible to register callbacks for each of those events.

Parameters:

Name	Type	Description	Default
`stream`	`ValueStreamHandler`	A handler for the matching data type.	required

subscribe(topic_selector: str)

Register the session to receive updates for the given topic.

Parameters:

Name	Type	Description	Default
`topic_selector`	`str`	The selector for topics to subscribe to.	required

unsubscribe `async` ¶

unsubscribe(topic_selector: str)

Unregister the session to stop receiving updates for the given topic.

Parameters:

Name	Type	Description	Default
`topic_selector`	`str`	The selector for topics to unsubscribe from.	required

remove_stream `async` ¶

remove_stream(stream: ValueStreamHandler)

Removes a stream.

Notes

More formally, self method removes all streams that compare equal to the given stream, regardless of the topic selector for which they are registered. It will also remove any fallback stream equal to the given stream. If there are no such streams, no changes are made.

Parameters:

Name	Type	Description	Default
`stream`	`ValueStreamHandler`	The value stream to remove.	required

diffusion.features.topics.details.topic_specification ¶

T_other `module-attribute` ¶

T_other = typing.TypeVar(
    "T_other", covariant=True, bound=ValueTypeProtocol
)

A DataType type variable.

WithProperties ¶

Bases: typing.Generic[T]

CompressionPolicy ¶

Bases: str, enum.Enum

Compression Level

OFF `class-attribute` `instance-attribute` ¶

OFF = 'off'

Disables compression completely for the topic and requires no additional CPU.

LOW `class-attribute` `instance-attribute` ¶

LOW = 'low'

Low compression.

Generally some compression is beneficial, so the default value for this property is this.

MEDIUM `class-attribute` `instance-attribute` ¶

MEDIUM = 'medium'

Medium compression.

HIGH `class-attribute` `instance-attribute` ¶

HIGH = 'high'

Compresses the topic messages to the smallest number of bytes, but has the highest CPU cost.

TopicDeliveryPriority ¶

Bases: str, enum.Enum

Topic Delivery Priority

LOW `class-attribute` `instance-attribute` ¶

LOW = 'low'

Low priority.

DEFAULT `class-attribute` `instance-attribute` ¶

DEFAULT = 'default'

Default priority.

HIGH `class-attribute` `instance-attribute` ¶

HIGH = 'high'

High priority.

ConflationPolicy ¶

Bases: str, enum.Enum

Conflation Policy

OFF `class-attribute` `instance-attribute` ¶

OFF = 'off'

Disables conflation for the topic. This policy disables all conflation for the topic, so topic updates will never be merged or discarded.

CONFLATE `class-attribute` `instance-attribute` ¶

CONFLATE = 'conflate'

Automatically conflates topic updates when back pressure is detected by the server.

UNSUBSCRIBE `class-attribute` `instance-attribute` ¶

UNSUBSCRIBE = 'unsubscribe'

Automatically unsubscribes the topic when back pressure is detected by the server. The unsubscription is not persisted to the cluster. If a session fails over to a different server it will be resubscribed to the topic.

ALWAYS `class-attribute` `instance-attribute` ¶

ALWAYS = 'always'

Automatically conflates topic updates as they are queued for the session. This is an eager policy that ensures only the latest update is queued for the topic, minimising the server memory and network bandwidth used by the session.

TopicSpecification ¶

Bases: typing.Generic[T], pydantic_v1_generics.GenericModel

Topic specifications provide the information required to create a topic. Topics can be created from a topic specification using add_topic.

Topic specifications allow an application to introspect the type and capabilities of a topic. Topic specifications are provided to ValueStream value streams and TopicNotifications topic notification listeners

A topic specification has a topic type and a map of property settings which define the behavior of the topic. A

Topic specification types can be derived from any datatype using its with_properties member. This will return a TopicSpecification type ready for instantiation with the relevant parameters.

Topic specifications with different properties can be derived from any instance using with_properties.

Topic Properties¶

Depending on the topic type, some properties must be included in the specification when creating a topic and some properties have no effect. The required and optional properties for each the topic type are set out in the following table. Properties unsupported by the topic type are ignored.

Default when optional	STRING JSON BINARY	DOUBLE INT64	RECORD_V2	TIME_SERIES
COMPRESSION	CompressionPolicy.LOW	Optional	—	Optional
CONFLATION	ConflationPolicy.CONFLATE	Optional	Optional	Optional
DONT_RETAIN_VALUE	`False`	Optional	Optional	Optional
OWNER	Optional	Optional	Optional	Optional
PERSISTENT	`True`	Optional	Optional	Optional
PRIORITY	TopicDeliveryPriority.DEFAULT	Optional	Optional	Optional
PUBLISH_VALUES_ONLY	`False`	Optional	—	Optional
REMOVAL	Optional	Optional	Optional	Optional
SCHEMA	—	—	Optional	—
TIDY_ON_UNSUBSCRIBE	`False`	Optional	Optional	Optional
TIME_SERIES_EVENT_VALUE_TYPE	—	—	—	Required
TIME_SERIES_RETAINED_RANGE	`limit 10`	—	—	—
TIME_SERIES_SUBSCRIPTION_RANGE	TIME_SERIES_SUBSCRIPTION_RANGE As documented
VALIDATE_VALUES	`False`	Optional	Optional	Optional

† TIME_SERIES topics have restricted values for the CONFLATION property. They are only allowed to have the values OFF or UNSUBSCRIBE.

topic_type `property` ¶

topic_type: typing.Type[T]

The topic datatype.

Returns:

Type	Description
`typing.Type[T]`	The topic datatype.

with_properties `class-attribute` ¶

with_properties: WithProperties = WithProperties()

Class property providing a TopicSpecification subtype ready for instantiation.

PUBLISH_VALUES_ONLY `class-attribute` `instance-attribute` ¶

PUBLISH_VALUES_ONLY: typing.Optional[bool] = None

Key of the topic property that specifies whether a topic should publish only values.

By default, a topic that supports delta streams will publish the difference between two values (a delta) when doing so is more efficient than publishing the complete new value. Subscribing sessions can use a ValueStream value stream to automatically apply the delta to a local copy of the topic value to calculate the new value.

Setting PUBLISH_VALUES_ONLY to True disables this behavior so that deltas are never published. Doing so is usually not recommended because it will result in more data being transmitted, less efficient use of network resources, and increased transmission latency. On the other hand, calculating deltas can require significant CPU from the server or, if update streams are used, from the updating client. The CPU cost will be higher if there are many differences between successive values, in which case delta streams confer fewer benefits. If successive values are unrelated to each other, consider setting PUBLISH_VALUES_ONLY to True. Also consider setting PUBLISH_VALUES_ONLY to True if the network capacity is high and the bandwidth savings of deltas are not required.

VALIDATE_VALUES `class-attribute` `instance-attribute` ¶

VALIDATE_VALUES: typing.Optional[bool] = None

Key of the topic property indicating whether a topic should validate inbound values.

By default, the server does not validate received values before sending them on to client sessions. Invalid or corrupt values will be stored in the topic and passed on to sessions. If this property is set to True, the server will perform additional validation on values to check that they are valid instances of the data type, and if it is not then it will return an error to the updater and not update the topic.

If this value is not set (or set to something other than True), no server validation of inbound values is performed. This is the recommended setting as there is a performance overhead to validation and a session using Topic Update cannot send invalid values anyway.

TIDY_ON_UNSUBSCRIBE `class-attribute` `instance-attribute` ¶

TIDY_ON_UNSUBSCRIBE: typing.Optional[bool] = None

Key of the topic property that specifies the 'tidy on unsubscribe' option for a topic.

By default, if a session unsubscribes from a topic, it will receive any updates for that topic that were previously queued but not sent.

If this property is set to "True", when a session unsubscribes from the topic, any updates for the topic that are still queued for the session are removed. There is a performance overhead to using this option as the client queue must be scanned to find topic updates to remove, however it may prove useful for preventing unwanted data being sent to sessions.

Since 6.8.3

SCHEMA `class-attribute` `instance-attribute` ¶

SCHEMA: typing.Optional[str] = None

Key of the topic property that specifies a schema which constrains topic values.

This property is only used by RECORD_V2 topics.

Since 6.8.3

DONT_RETAIN_VALUE `class-attribute` `instance-attribute` ¶

DONT_RETAIN_VALUE: typing.Optional[bool] = None

Key of the topic property that specifies a topic should not retain its last value.

By default, a topic will retain its latest value. The latest value will be sent to new subscribers. Setting this property to True disables this behavior. New subscribers will not be sent an initial value. No value will be returned for fetch operations that select the topic. This is useful for data streams where the values are only transiently valid.

Setting DONT_RETAIN_VALUE to True also disables delta streams, regardless of the PUBLISH_VALUES_ONLY value. If subsequent values are likely to be related, delta streams usually provide performance benefits (see PUBLISH_VALUES_ONLY). Consider leaving DONT_RETAIN_VALUE set to False to benefit from delta streams, even if there is no other requirement to retain the last value.

Bearing in mind the performance trade-offs of disabling delta streams, there are two reasons to consider setting DONT_RETAIN_VALUE to True. First, it stops the server and each subscribed client from keeping a copy of the value, reducing their memory requirements. Second, when a topic has a high update rate and is replicated across a cluster, it can significantly improve throughput because the values need not be persisted to the cluster.

Time series topics ignore this property and always retain the latest value.

Since 6.8.3

PERSISTENT `class-attribute` `instance-attribute` ¶

PERSISTENT: typing.Optional[bool] = None

Key of the topic property that can be used to prevent a topic from being persisted when the server is configured to enable persistence.

By default, a topic will be persisted if persistence is enabled at the server and the topic type supports persistence. Setting PERSISTENT to `false` will prevent the topic from being persisted. Since 6.8.3

REMOVAL `class-attribute` `instance-attribute` ¶

REMOVAL: typing.Optional[str] = None

Key of the topic property that specifies a removal policy for automatic removal of the topic (and/or other topics). This property is specified as an expression which defines one or more conditions that are to be satisfied before automatic removal occurs. The expression takes the form: when conditions [remove "selector"] At least one condition must be supplied. If more than one is supplied, they must be separated by logical operators (and or or). The natural evaluation order of the operators may be changed by surrounding with parentheses (e.g. (condition and condition)). The remove clause is optional. It provides a Topic Selector expression representing the topics to be removed. If a remove clause is specified, the topic with the removal policy will only be removed if its path matches the selector expression. The selector must be surrounded by either double or single quotes. When many topics have the same removal policy, it is better to set the REMOVAL property for one of them, using a remove clause that selects all of the topics. This is more efficient because it allows the server to avoid evaluating the same condition many times.

The permissions that are applied at the time of removal are those defined by the roles of the principal that created the topic at the time of creation. The roles of that principal may therefore change before the removal with no effect, but if the permissions given to the roles change it may have an effect upon the final removal. Only one occurrence of each of the following condition types may be included within the expression:

Condition Type	Format	Usage
time after	`time after absoluteTime`	Removal should occur after a specified absolute time. Absolute time may be specified as a number of milliseconds since the epoch (00:00:00 on 1 January 1970) or as a quoted date and time formatted in RFC_1123 date time format. Either single or double quotes may be used.
subscriptions less than	`[local] subscriptions < n for forPeriod [after afterPeriod]`	Removal should occur when the topic has had less than the specified number (n) of subscriptions for a given period (forPeriod) of time. Optionally, an initial period (afterPeriod) may be specified by which to delay the initial checking of this condition. See below for period formats. The optional `local` keyword restricts evaluation to only count subscriptions from sessions belonging to the local server or cluster, ignoring subscriptions from sessions belonging to downstream remote servers that host fanout replicas of the topic.
no updates for	`no updates for forPeriod [after afterPeriod]`	Removal should occur when the topic has had no updates for a given period (forPeriod) of time. Optionally, an initial period (afterPeriod) may be specified by which to delay the initial checking of this condition. See below for period formats.

Multiple occurrences of the following condition types may be included within the expression:

Condition Type	Format	Usage
no session has	no [local] session has "criteria" [for forPeriod] [after afterPeriod]	Removal should occur when there are no sessions satisfying certain criteria. Optionally the criteria can be required to be satisfied for a period of time (forPeriod). Optionally, an initial period (afterPeriod) can be specified to delay the initial check of the criteria. Session selection criteria are specified as defined in session filters and must be surrounded by single or double quotes. See below for period formats. The optional local keyword restricts evaluation to sessions belonging to the local server or cluster, ignoring sessions belonging to downstream remote servers that host fanout replicas of the topic.
this session closes	This is a shorthand form of `no local session has` that may be used to indicate that the topic is to be removed when the session that created it closes.

Time periods are specified as a number followed (with no intermediate space) by a single letter representing the time unit. The time unit may be s (seconds), m (minutes), h (hours) or d (days). For example, 10 minutes would be specified as 10m. If quotes or backslashes \ are required within quoted values such as selectors or session criteria then they may be escaped by preceding with \. The expression is validated only by the server and therefore if an invalid expression is specified it will be reported as an InvalidTopicSpecificationError.

** Examples: **

when time after 1518780068112

The topic will be removed when the date and time indicated by the specified number of milliseconds since the epoch has passed.

when time after "Tue, 3 Jun 2018 11:05:30 GMT"

The topic will be removed when the specified date and time has passed.

when time after "Tue, 3 Jun 2018 11:05:30 GMT" remove "*alpha/beta//"

The topic alpha/beta and all topics subordinate to it will be removed when the specified date and time has passed.

when subscriptions < 1 for 20m

The topic will be removed when it has had no subscriptions for a continuous period of 20 minutes.

when subscriptions < 2 for 20m after 1h

The topic will be removed when it has had less than 2 subscriptions for a continuous period of 20 minutes after one hour has passed since its creation.

when no updates for 3h

The topic will be removed when it has had no updates for a continuous period of 3 hours.

when no updates for 15m after 1d

The topic will be removed when it has had no updates for a continuous period of 15 minutes after one day has passed since its creation.

when this session closes

The topic will be removed when the session creating it closes.

when no session has '$Principal is "Alice"'

The topic will be removed when there are no sessions with the principal 'Alice'.

when no session has '$Principal is "Alice"' for 10m

The topic will be removed when there are no sessions with the principal 'Alice' for a continuous period of 10 minutes.

when no session has 'Department is "Accounts"' for 30m after 2h

The topic will be removed when there have been no sessions from the Accounts department for a continuous period of 30 minutes after 2 hours have passed since its creation.

when time after "Tue, 3 Jun 2018 11:05:30 GMT" and subscriptions < 1 for 30m

The topic will be removed when the specified date and time has passed and the topic has had no subscriptions for a continuous period of 30 minutes after that time.

when time after "Tue, 3 Jun 2018 11:05:30 GMT" and subscriptions < 2 for 10m after 1h

The topic will be removed when the specified date and time has passed and the topic has had less than 2 subscriptions for a continuous period of 10 minutes after that time plus one hour.

when time after "Tue, 3 Jun 2018 11:05:30 GMT" or subscriptions < 2 for 10m after 1h

The topic will be removed when the specified date and time has passed or the topic has had less than 2 subscriptions for a continuous period of 10 minutes after one hour from its creation.

when time after "Tue, 3 Jun 2018 11:05:30 GMT" and (subscriptions < 2 for 10m after 1h or no updates for 20m)

The topic will be removed when the specified date and time has passed and either the topic has had less than 2 subscriptions for a continuous period of 10 minutes after that time plus one hour or it has had no updates for a continuous period of 20 minutes. Note that the parentheses are significant here as without them the topic would be removed if it had had no updates for 20 minutes regardless of the time and subscriptions clause.

Notes and restrictions on use

The after time periods refer to the period since the topic was created or restored from persistence store after a server is restarted. They are designed as a 'grace' period after the topic comes into existence before the related condition starts to be evaluated. When not specified the conditions start to be evaluated as soon as the topic is created or restored. The server will evaluate conditions on a periodic basis (every few seconds) so the exact removal time will not be precise for low periodic granularity. The meaning of the for period in a no session has condition is subtly different from its use in other conditions. It does not guarantee that there has been no session satisfying the condition at some point between evaluations, only that when evaluated the given period of time has passed since it was last evaluated and found to have no matching sessions. Subscriptions is the number of subscriptions to a topic. Automatic topic removal is supported for a topic that is replicated across the local cluster, and for a topic with with fanout replicas on downstream remote servers. A subscriptions less than condition will be evaluated against the total number of subscriptions across the cluster and on all fanout replicas on downstream remote servers. A no session has condition will consider all sessions hosted across the cluster and all sessions hosted by downstream remote servers that have a fanout replica of the topic. The local keyword can be used to restrict evaluation to the local cluster, ignoring fanout replicas.

Since 6.8.3

CONFLATION `class-attribute` `instance-attribute` ¶

CONFLATION: typing.Optional[ConflationPolicy] = None

The conflation policy of the topic. The policy specifies how the server manages queued topic updates. Conflation is applied individually to each session queue. Conflation is the process of merging or discarding topic updates queued for a session to reduce the server memory footprint and network data. The server will conflate sessions that have a large number of queued messages to meet configured queue size targets. The sessions with the largest queues are typically slow consumers or have been disconnected – both will benefit from conflation. This property allows conflation behavior to be tuned on a topic-by-topic basis.

The supported policies are:

OFF
CONFLATE
UNSUBSCRIBE
ALWAYS

The default policy used when the property is not specified and the topic type is not time series is CONFLATE. The default policy used when the property is not specified and the topic type is time series is OFF.

See ConflationPolicy for more.

The CONFLATE and UNSUBSCRIBE policies are applied when the server detects back pressure for a session. The server configuration places limits on the data queued for each session. If these limits are breached, the server will conflate the session queue to attempt to reducee its size. If the session queue still exceeds the limits after conflation, the session will be terminated. Conflation can be disabled on a session-by-session basis. If conflation is disabled for a session the policy will not be applied to topic updates queued for the session but will be for other sessions that have conflation enabled. The policies CONFLATE and ALWAYS are not supported for time series topics as they would cause missing events. Attempts to enable these policies with time series topics will cause the creation of the topic to fail, reporting that the specification is invalid.

Since 6.8.3

OWNER `class-attribute` `instance-attribute` ¶

OWNER: typing.Optional[str] = None

Key of the topic property that allows the creator of a topic to extend READ_TOPIC, MODIFY_TOPIC, and UPDATE_TOPIC permissions to a specific principal, in addition to the permissions granted by the authorisation rules in the security store. A session that has authenticated using the principal can update and remove the topic, so the principal can be considered the topic owner. To fetch or subscribe to the topic, the principal must also be granted the SELECT_TOPIC permission by the security store rules. This may be used in the following cases:
1) A session creates a topic and makes its own principal the owner.
2) A session creates a topic and makes another principal the owner. The format of the property value is: $Principal is name where name is the name of the principal. Single quotes may be used instead of double quotes and special characters can be escaped. The purpose of this property is to allow a client to create topics on behalf of other users. This can be used in conjunction with the REMOVAL property so that such topics are removed when there are no longer any sessions for the named principal. For example:

>>> import diffusion.datatypes
>>> specification = diffusion.datatypes.JSON
>>> specification.with_properties(OWNER="$Principal is 'myPrincipal'", REMOVAL="when no session has '$Principal is "myPrincipal"' for 5s")

Since 6.8.3

COMPRESSION `class-attribute` `instance-attribute` ¶

COMPRESSION: typing.Optional[
    typing.Union[CompressionPolicy, bool]
] = None

Key of the topic property that allows the compression policy to be set on a per-topic basis.

Compression reduces the bandwidth required to broadcast topic updates to subscribed sessions, at the cost of increased server CPU.

Changes to a topic's value are published to each subscribed session as a sequence of topic messages. A topic message can carry the latest value or the difference between the latest value and the previous value (a delta). The compression policy determines if and how published topic messages are compressed. Topic messages are not exposed through the client API; the client library handles decompression and decodes deltas automatically, passing reconstructed values to the application.

The compression policy for a topic is specified by setting this property to one of several values:

OFF
LOW
MEDIUM
HIGH

The policies are listed in the order of increasing compression and increasing CPU cost.

PRIORITY `class-attribute` `instance-attribute` ¶

PRIORITY: typing.Optional[TopicDeliveryPriority] = None

Key of the topic property that specifies the topic delivery priority. The supported delivery priorities are:

LOW
DEFAULT
HIGH

The delivery priority affects the order of topic updates sent to a subscribed client session. When there are multiple topic updates for topics with different priorities in a session's outbound queue, updates for HIGH priority topics will be delivered first, followed by updates for DEFAULT priority topics, followed by updates for LOW priority topics. Topic subscription and unsubscription notifications are also delivered according to the topic delivery priority.

Using different delivery priorities is most beneficial when there is a large backlog of queued updates to deliver to a client session. On lightly loaded systems, updates typically remain in the outbound queue for a few milliseconds and so there is a lower chance of topic updates being reordered based on their priority. The backlog will be larger if the topic update rate is higher; the server or the client are more heavily loaded; the client session becomes temporarily disconnected; or if there is poor network connectivity between the server and the client.

Messages from the server to the client that are not topic updates, for example ping requests and responses, are queued with the DEFAULT delivery priority.

Since 6.8.3

new_topic_specification `classmethod` ¶

new_topic_specification(
    tp: typing.Type[T_other],
) -> typing.Type[TopicSpecification[T_other]]

Return a TopicSpecification class with datatype tp, ready for instantiation.

This is for internal use - to get a TopicSpecification for a given DataType you should use its with_properties member, which will return an appropriate TopicSpecification type ready for instantiation with the relevant parameters.

Parameters:

Name	Type	Description	Default
`tp`	`typing.Type[T_other]`	the datatype	required

Returns: A new TopicSpecification with the given type

diffusion.features.topics.streams ¶

Stream handlers for topics.

LOG `module-attribute` ¶

LOG = structlog.get_logger()

diffusion.handlers.SubHandlerProtocol or subclass.

SubHandlerValues ¶

Bases: typing_extensions.TypedDict, typing.Generic[AT_T_contra, Callback_RT]

Handlers that can be passed into the ValueStreamHandler

unsubscribe `instance-attribute` ¶

unsubscribe: Annotated[
    typing.Optional[
        UnsubscribeProtocol[AT_T_contra, Callback_RT]
    ],
    Doc("An unsubscribe handler."),
]

An unsubscribe handler.

subscribe: Annotated[
    typing.Optional[
        SubHandlerProtocol[AT_T_contra, Callback_RT]
    ],
    Doc("A subscribe handler."),
]

A subscribe handler.

update `instance-attribute` ¶

update: Annotated[
    typing.Optional[
        UpdateProtocol[AT_T_contra, Callback_RT]
    ],
    Doc("An update handler."),
]

An update handler.

close `instance-attribute` ¶

close: Annotated[
    typing.Optional[
        SubHandlerProtocol[AT_T_contra, Callback_RT]
    ],
    Doc("A close handler."),
]

A close handler.

ValueStreamHandler ¶

Bases: EventStreamHandler[AT_T_co, Callback_RT]

Stream handler implementation for the value streams of the given type.

init ¶

__init__(
    data_type: Annotated[
        Type[AT_T_co],
        Doc("the data type associated with this handler"),
    ],
    **kwargs: Unpack[SubHandlerValues[AT_T_co, Callback_RT]]
) -> None

Initialise the ValueStreamHandler

Parameters:

Name	Type	Description	Default
`data_type`	`Annotated[Type[AT_T_co], Doc('the data type associated with this handler')]`	the data type associated with this handler	required

Other Parameters:

Name	Type	Description
`unsubscribe`	`Annotated[typing.Optional[UnsubscribeProtocol[AT_T_contra, Callback_RT]], Doc('An unsubscribe handler.')]`	An unsubscribe handler.
`subscribe`	`Annotated[typing.Optional[SubHandlerProtocol[AT_T_contra, Callback_RT]], Doc('A subscribe handler.')]`	A subscribe handler.
`update`	`Annotated[typing.Optional[UpdateProtocol[AT_T_contra, Callback_RT]], Doc('An update handler.')]`	An update handler.
`close`	`Annotated[typing.Optional[SubHandlerProtocol[AT_T_contra, Callback_RT]], Doc('A close handler.')]`	A close handler.

Parameters:

Name	Type	Description	Default
`data_type`	`Annotated[Type[AT_T_co], Doc('the data type associated with this handler')]`	the data type associated with this handler	required
`**kwargs`	`Unpack[SubHandlerValues[AT_T_co, Callback_RT]]`	as documented in "Other Parameters."	`{}`

diffusion.features.topics.selectors ¶

Matching of topic selectors.

Selector ¶

Bases: Protocol, Hashable

Generic protocol for all selectors.

raw `property` ¶

raw: str

The original, unoptimised value of the selector.

match ¶

match(topic_path: str) -> bool

Compares a topic path against the selector.

Parameters:

Name	Type	Description	Default
`topic_path`	`str`	The topic path to match against the selector.	required

Returns:

Type	Description
`bool`	`True` if the path matches the selector; `False` otherwise.

TopicSelectorError ¶

Bases: DiffusionError

Error related to selector matching.

PathSelector ¶

Bases: Selector

Basic topic selector, matching the topic path exactly.

Parameters:

Name	Type	Description	Default
`selector`	`str`	The topic selector text.	required

raw `property` ¶

raw: str

The original, unoptimised value of the selector.

match ¶

match(topic_path: str) -> bool

Compares a topic path against the selector.

Parameters:

Name	Type	Description	Default
`topic_path`	`str`	The topic path to match against the selector.	required

Returns:

Type	Description
`bool`	`True` if the path matches the selector; `False` otherwise.

PatternSelector ¶

Bases: PathSelector

Common functionality of pattern selectors.

FullPathPatternSelector ¶

Bases: PatternSelector

Selector containing a regular expression matching the entire path.

A full-path pattern topic selector returns topics for which the regular expression matches the full topic path.

Parameters:

Name	Type	Description	Default
`selector`	`str`	The topic selector text.	required

match ¶

match(topic_path: str) -> bool

Compares a topic path against the selector.

Parameters:

Name	Type	Description	Default
`topic_path`	`str`	The topic path to match against the selector.	required

Returns:

Type	Description
`bool`	`True` if the path matches the selector; `False` otherwise.

SplitPathPatternSelector ¶

Bases: PatternSelector

Selector containing split-path regular expression pattern.

A split-path pattern expression contains a list of regular expressions separated by the / character. Each regular expression describes a part of the topic path. The selector returns topics for which each regular expression matches the part of the topic path at the corresponding level.

Parameters:

Name	Type	Description	Default
`selector`	`str`	The topic selector text.	required

match ¶

match(topic_path: str) -> bool

Compares a topic path against the selector.

Parameters:

Name	Type	Description	Default
`topic_path`	`str`	The topic path to match against the selector.	required

Returns:

Type	Description
`bool`	`True` if the path matches the selector; `False` otherwise.

SelectorSet ¶

Bases: Set[Selector], Selector

A set of multiple path or pattern selectors.

A selector set returns topics that match any of the selectors. Its textual form is a list of selectors separated by the separator ////.

Parameters:

Name	Type	Description	Default
`selectors`	`Union[str, Iterable[str]]`	Either the full list of selectors (separated by `////`) or an iterable of individual selectors.	required

raw `property` ¶

raw: str

The original, unoptimised value of the selector.

match ¶

match(topic_path: str) -> bool

Compares a topic path against the selector.

Parameters:

Name	Type	Description	Default
`topic_path`	`str`	The topic path to match against the selector.	required

Returns:

Type	Description
`bool`	`True` if the path matches the selector; `False` otherwise.

any_of `classmethod` ¶

any_of(*args: str) -> SelectorSet

Helper method that accepts selectors as individual arguments to create a set.

get_selector ¶

get_selector(selector: str) -> Selector

Instantiate a selector based on the given textual value.

diffusion.features.topics.update ¶

UpdateConstraint ¶

Bases: abc.ABC, GenericModel

The constraint to be applied to an update operation or the creation of an update stream.

Notes

Constraints describe a condition that must be satisfied for an operation to succeed. Constraints can be applied to the setting of a value or creation of an update stream. Constraints are only evaluated on the server.

The constraints are evaluated using the:

active session locks
existence of the topic
current value of the topic

The value of a topic can be described in several ways. The value can be described as an exact value, a partial value or an unset value.

Constraints can be composed with one another. It is only possible to construct logical ANDs of constraints. Constraints can only be composed if the resulting constraint is satisfiable. Multiple session locks can be held but a topic can only have a single value. Constraints specifying multiple topic values can't be constructed.

Constraints can be created using a ConstraintFactory.

and ¶

__and__(other: UpdateConstraint) -> UpdateConstraint

Creates a composed constraint that represents a logical AND of this constraint and another.

Notes

You must use the binary & form to invoke this due to short-circuiting.

Parameters:

Name	Type	Description	Default
`other`	`UpdateConstraint`	The constraint that will be logically-ANDed with the current constraint.	required

Returns:

Type	Description
`UpdateConstraint`	The newly composed constraint that represents a logical AND of the current constraint and the `other` constraint.

constraint_factory ¶

ConstraintFactory ¶

Bases: abc.ABC

The factory for the update constraint types.

json_value `property` ¶

json_value: PartialJSON

Gets the constraint that partially matches the current topic value.

Notes

The topic must be a diffusion.datatypes.JSON topic. The diffusion.features.topics.update.constraints.PartialJSON partially describes the structure of a diffusion.datatypes.JSON value.

Returns:

Type	Description
`PartialJSON`	The constraint.

no_value `property` ¶

no_value: NoValueConstraint

Gets the constraint requiring the topic to have no value.

Notes

This is useful when setting the first value of a topic. This constraint is unsatisfied if no topic is present at the path, making it unsuitable for operations that try to add topics.

Returns:

Type	Description
`NoValueConstraint`	The constraint.

no_topic `property` ¶

no_topic: NoTopicConstraint

Gets the constraint requiring the path to have no topic.

Notes

This is useful when setting the first value of a topic being added using Topics.add_and_set_topic without changing the value if the topic already exists. This constraint is unsatisfied if a topic is present at the path, making it unsuitable for operations that try to set topics without adding them.

Returns:

Type	Description
`NoTopicConstraint`	The constraint.

value ¶

value(
    value: typing.Union[AbstractDataType, bytes],
) -> BinaryValueConstraint

Creates a constraint requiring the current value of the topic to match the supplied value.

Notes

When a None value, the topic is set to have no value. Use the NoValueConstraint constraint to check if the topic has no value.

This is useful when changing the value of a topic. This constraint is unsatisfied if no topic is present at the path, making it unsuitable for operations that try to add topics.

Parameters:

Name	Type	Description	Default
`value`	`typing.Union[AbstractDataType, bytes]`	The value.	required

Returns:

Type	Description
`BinaryValueConstraint`	The update constraint.

constraints ¶

PartialJSONType `module-attribute` ¶

PartialJSONType = typing.TypeVar('PartialJSONType')

Instance of PartialJSON or any subclasses

Unconstrained ¶

Bases: UpdateConstraint

The unconstrained update constraint.

BinaryValueConstraint ¶

Bases: UpdateConstraint

The update constraint that requires the exact binary value.

NoValueConstraint ¶

Bases: UpdateConstraint

The update constraint that requires a topic to have no value.

LockConstraint ¶

Bases: UpdateConstraint

The constraint on a SessionLock

NoTopicConstraint ¶

Bases: UpdateConstraint

The update constraint that requires the path to have no topic.

PartialJSON ¶

Bases: UpdateConstraint

The constraint requiring the current value of the JSON topic to match the partially described value.

with_ ¶

with_(
    pointer: StrictStr, value: AbstractDataType
) -> "PartialJSONType"

Requires a value at a specific position in the JSON object.

Notes

The pointer is a JSON Pointer (https://tools.ietf.org/html/rfc6901) syntax reference locating the value in the JSON object.

The pointer syntax is not being verified for correctness.

Parameters:

Name	Type	Description	Default
`pointer`	`StrictStr`	The pointer expression.	required
`value`	`AbstractDataType`	The value.	required

Returns:

Type	Description
`'PartialJSONType'`	The constraint including the specified `pointer`.

without ¶

without(pointer: StrictStr) -> 'PartialJSONType'

Requires a specific position in the JSON object to be empty.

Notes

The pointer is a JSON Pointer (https://tools.ietf.org/html/rfc6901) syntax reference that should have no value in the JSON object.

The pointer syntax is not being verified for correctness.

Parameters:

Name	Type	Description	Default
`pointer`	`StrictStr`	The pointer expression.	required

Returns:

Type	Description
`'PartialJSONType'`	The constraint including the specified `pointer`.

diffusion.features.topics.fetch ¶

fetch_common ¶

IVoidFetch ¶

Bases: AbstractDataType[TopicSpecification['IVoidFetch'], NoneType, NoneType]

Internal representation of an untyped fetch request/result.

encode `classmethod` ¶

encode(value: Any) -> bytes

Convert a value into the corresponding binary representation.

Parameters:

Name	Type	Description	Default
`value`	`Any`	Native value to be serialised	required

Returns:

Type	Description
`bytes`	Serialised binary representation of the value.

decode `classmethod` ¶

decode(data: bytes) -> Any

Convert a binary representation into the corresponding value.

Parameters:

Name	Type	Description	Default
`data`	`bytes`	Serialised binary representation of the value.	required

Returns:

Type	Description
`Any`	Deserialised value.

fetch_query ¶

Limit ¶

Bases: GenericModel

The range limit for a fetch request.

LimitExisting ¶

Bases: Limit

path `class-attribute` `instance-attribute` ¶

path: str = dc.field()

The path.

includes_path `class-attribute` `instance-attribute` ¶

includes_path: bool = dc.field()

Whether the path is included or not.

FetchRange ¶

Bases: GenericModel

The fetch range.

FetchQuery ¶

Bases: GenericModel

The fetch request.

selector `class-attribute` `instance-attribute` ¶

selector: pydantic.StrictStr = dc.field()

The topic selector.

range `class-attribute` `instance-attribute` ¶

range: FetchRange = dc.field()

The fetch range.

topic_types `class-attribute` `instance-attribute` ¶

topic_types: typing.FrozenSet[
    typing.Type[FetchQueryValueType]
] = dc.field()

The topic types.

with_values `class-attribute` `instance-attribute` ¶

with_values: pydantic.StrictBool = dc.field()

Whether values are required.

with_properties `class-attribute` `instance-attribute` ¶

with_properties: pydantic.StrictBool = dc.field()

Whether properties are required.

limit `class-attribute` `instance-attribute` ¶

limit: SignedInt32 = SignedInt32(0)

Optional limit. 0 = none. Positive = first, negative = last.

maximum_result_size `class-attribute` `instance-attribute` ¶

maximum_result_size: MaximumResultSize = MaximumResultSize(
    0
)

Optional maximum result size. 0 = taken from server maximum message size. If specified must be >= 1024.

deep_branch_depth `class-attribute` `instance-attribute` ¶

deep_branch_depth: NonNegativeSignedInt32 = (
    NonNegativeSignedInt32.max()
)

The minimum number of parts in a path that belongs to a "deep" branch. NonNegativeSignedInt32.max if the query is not depth limited.

deep_branch_limit `class-attribute` `instance-attribute` ¶

deep_branch_limit: NonNegativeSignedInt32 = (
    NonNegativeSignedInt32.max()
)

The maximum number of paths to return for each deep branch. Int32.MAX_VALUE if the query is not depth limited.

with_unpublished_delayed_topics `class-attribute` `instance-attribute` ¶

with_unpublished_delayed_topics: pydantic.StrictBool = (
    dc.field(default=False)
)

Whether unpublished delayed topics should be included in the results.

FetchQuerySerializer ¶

Bases: GenericConfig[FetchQuery]

The FetchQuery serializer.

fetch_query_result ¶

FetchTopicResult ¶

Bases: typing.Generic[A_T], GenericModel

The single topic result from a FetchQuery.

FetchQueryResult ¶

Bases: GenericModel

The result from a FetchQuery.

fetch_request ¶

FetchContext `dataclass` ¶

session `instance-attribute` ¶

session: InternalSession

The internal session

maximum_message_size `instance-attribute` ¶

maximum_message_size: StrictNonNegativeInt

The maximum message size.

FetchRequest ¶

Bases: typing.Generic[TValue], GenericModel

The generic version of FetchResult that is typed to a particular value type.

Notes: This defines a fetch request to be made to the server. A request can be created using the FetchRequest property and modified to specify a range of topics and/or various levels of detail.

The fetch request is issued to the server using the fetch method supplying a topic selector which specifies the selection of topics.

A range is defined as being between two points of the topic tree which is ordered in full path name order. So an example tree order would be:

a
a/b
a/c
a/c/x
a/c/y
a/d
a/e
b
b/a/x
b/b/x
c

The start point of a range can be specified using from_ or after and an end point using to or before. from_ and to include any topic with the specified path in the selection, whereas after and before are non-inclusive and useful for paging through a potentially large range of topics. If no start point is specified, the start point is assumed to be the logical beginning of the topic tree. Similarly, if no end point is specified, the end point is the logical end of the topic tree. Ranges should be within the scope indicated by the topic selector used when issuing the fetch.

As a minimum, the path and type of each topic selected will be returned. It is also possible to request that the topic values and/or properties are returned.

If values are selected then the topic types selected are naturally constrained by the value class indicated. So if OBJECT will return values for all topic types.

To select topic types when values are not required, or to further constrain the selection when values are required, it is also possible to specify exactly which topic types to select.

A limit on the number of results returned can be specified using first. This is advisable when the result set could potentially be large. When such a limit is used then the result will indicate whether more results for the same selection would be available via the FetchResult.has_more property.

The request can be sent to the server using fetch and the results are returned in path order, earliest path first, starting from the beginning of any range specified.

It is also possible to request results from the end of the range indicated by specifying a limit to the number of results using last. This returns up to the specified number of results from the end of the range, in path order. This is useful for paging backwards through a range of topics.

If values are requested, the request and results are typed to the value type (FetchRequest[TValue]) and FetchResult[TValue] respectively), otherwise the non-generic versions of FetchRequest and FetchResult are being used.

ROUTING topics are not supported, and if encountered will be ignored (i.e. treated as if they did not exist). *

FetchRequest[TValue] instances are immutable and can be safely shared and reused.

Added in version 6.11.

from_ ¶

from_(topic_path: StrictStr) -> FetchRequest[TValue]

Specifies a logical start point within the topic tree.

Notes

If specified, only results for topics with a path that is lexically equal to or 'after' the specified path will be returned.

This is the inclusive equivalent of after and if used will override any previous after or from_ constraint.

Parameters:

Name	Type	Description	Default
`topic_path`	`StrictStr`	The topic path from which results are to be returned.	required

Returns:

Type	Description
`FetchRequest[TValue]`	The fetch request derived from this fetch request but selecting only topics from the specified path onwards (inclusive).

after ¶

after(topic_path: StrictStr) -> FetchRequest[TValue]

Specifies a logical start point within the topic tree.

Notes

If specified, only results for topics with a path that is lexically 'after' the specified path will be returned.

This is the non-inclusive equivalent of from_ and if used will override any previous from_ or after constraint.

Parameters:

Name	Type	Description	Default
`topic_path`	`StrictStr`	The topic path after which results are to be returned.	required

Returns:

Type	Description
`FetchRequest[TValue]`	The fetch request derived from this fetch request but selecting only topics after the specified path (not inclusive).

to ¶

to(topic_path: StrictStr) -> FetchRequest[TValue]

Specifies a logical end point within the topic tree.

Notes

If specified, only results for topics with a path that is lexically equal to or 'before' the specified path will be returned.

This is the inclusive equivalent of before and if used will override any previous before or to constraint.

Parameters:

Name	Type	Description	Default
`topic_path`	`StrictStr`	The topic path before which results are to be returned.	required

Returns:

Type	Description
`FetchRequest[TValue]`	The fetch request derived from this fetch request but selecting only topics including and before the specified path.

before ¶

before(topic_path: StrictStr) -> FetchRequest[TValue]

Specifies a logical end point within the topic tree.

Notes

If specified, only results for topics with a path that is lexically 'before' the specified path will be returned.

This is the non-inclusive equivalent of to and if used will override any previous to or before constraint.

Parameters:

Name	Type	Description	Default
`topic_path`	`StrictStr`	The topic path before which results are to be returned.	required

Returns:

Type	Description
`FetchRequest[TValue]`	The fetch request derived from this fetch request but selecting only topics before the specified path (not inclusive).

topic_types ¶

topic_types(topic_types: TopicTypeSet) -> FetchSelf

Specifies that only topics of the specified topic types should be returned.

Notes

If this is not specified, all types will be returned (unless constrained by with_values).

If the specified topic type matches the event type of a time series topic it will also be returned. The value will be delivered without the associated metadata. To specify all time series topics use diffusion.datatypes.TIME_SERIES.

This may be used instead to further constrain the results when using with_values. For example, you can specify JSON to with_values then specify diffusion.datatypes.JSON here to ensure that only JSON topics are returned and not those topics that are logically value subtypes of JSON (e.g. diffusion.datatypes.STRING).

If with_values has been specified then the types specified here must be compatible with the specified value type or the event type for time series topics.

ROUTING may not be specified as only target topic types may be selected.

Parameters:

Name	Type	Description	Default
`topic_types`	`TopicTypeSet`	The topic types to be selected.	required

Returns:

Type	Description
`FetchSelf`	The fetch request derived from this fetch request but specifying that only topics of the specified topic types should be returned.

with_values ¶

with_values() -> FetchRequest[IVoidFetch]

with_values(
    tp_new: typing.Type[TNewValue_Not_IVoidFetch],
) -> FetchRequest[TNewValue_Not_IVoidFetch]

with_values(
    tp_new: typing.Union[
        typing.Type[IVoidFetch],
        typing.Type[TNewValue_Not_IVoidFetch],
    ] = IVoidFetch,
) -> typing.Union[
    FetchRequest[IVoidFetch],
    FetchRequest[TNewValue_Not_IVoidFetch],
]

Specifies that values should be returned for selected topics.

Parameters:

Name	Type	Description	Default
`tp_new`	`typing.Union[typing.Type[IVoidFetch], typing.Type[TNewValue_Not_IVoidFetch]]`	the type of values the fetch request should return.	`IVoidFetch`

Notes

This constrains the selection to only those topics with a data type compatible with the specified TNewValue

If OBJECT is specified, values will be returned for all topic types (unless constrained by topic_types).

The specified value constraints the topic types. So, any topic types specified in a previous call to topic_types that cannot be read as the specified class will be removed from the list of topic types.

Returns:

Type	Description
`typing.Union[FetchRequest[IVoidFetch], FetchRequest[TNewValue_Not_IVoidFetch]]`	The fetch request derived from this fetch request but specifying that only topics compatible with the specified type should be returned with values.

without_values ¶

without_values() -> FetchRequest[IVoidFetch]

Specifies that no values should be returned for selected topics.

Notes

This methods lifts previous data type constraints set by with_values.

Returns:

Type	Description
`FetchRequest[IVoidFetch]`	The fetch request derived from this fetch request but specifying that topics should not be returned with values.

with_properties ¶

with_properties() -> FetchRequest[TValue]

Specifies that all properties associated with each topic's ITopicSpecification should be returned.

Returns:

Type	Description
`FetchRequest[TValue]`	The fetch request derived from this fetch request but specifying that topic specification properties should be returned.

first ¶

first(number: StrictNonNegativeInt) -> FetchSelf

Specifies a maximum number of topic results to be returned from_ the start of the required range.

Notes

If this is not specified, the number of results returned will only be limited by other constraints of the request.

This should be used to retrieve results in manageable batches and prevent very large result sets.

If there are potentially more results that would satisfy the other constraints then the fetch result will indicate so via the FetchResult.has_more property.

Zero can be supplied to return no results. Such a request can be used together with FetchResult.has_more to query whether there are topics that match the selector provided to fetch, without retrieving the details of any of the topics. To retrieve unlimited topics use Int32.MaxValue which is the default value.

Either this or last may be specified. This will therefore override any previous last or first constraint.

Parameters:

Name	Type	Description	Default
`number`	`StrictNonNegativeInt`	The maximum number of results to return from the start of the range.	required

Returns:

Type	Description
`FetchSelf`	The fetch request derived from this fetch request but selecting only the number of topics specified from the start of the range.

last ¶

last(number: StrictNonNegativeInt) -> FetchSelf

Specifies a maximum number of topic results to be returned from_ the end of the required range.

Notes

This is similar to first except that the specified number of results are returned from_ the end of the range. This is useful for paging backwards through a range of topics. The results will be returned in topic path order (not reverse order).

Zero can be supplied to return no results. Such a request can be used together with FetchResult.has_more to query whether there are topics that match the selector provided to fetch, without retrieving the details of any of the topics. To retrieve unlimited topics use Int32.MaxValue which is the default value.

Either this or first may be specified. This will therefore override any previous first or last constraint.

Parameters:

Name	Type	Description	Default
`number`	`StrictNonNegativeInt`	The maximum number of results to return from the end of the range.	required

Returns:

Type	Description
`FetchSelf`	The fetch request derived from this fetch request but selecting only the number of topics specified from the end of the range.

maximum_result_size ¶

maximum_result_size(
    maximum: StrictNonNegativeInt,
) -> FetchSelf

Specifies the maximum data size of the result set.

Notes

This may be used to constrain the size of the result. If not specified then by default the maximum message size for the session (as specified by maximum_message_size is used.

If a value greater than the session's maximum message size then the maximum message size will be used.

Parameters:

Name	Type	Description	Default
`maximum`	`StrictNonNegativeInt`	The maximum size of the result set in bytes.	required

Returns:

Type	Description
`FetchSelf`	The fetch request derived from this fetch request but constraining the size of the result to the sspecified `maximum`.

limit_deep_branches ¶

limit_deep_branches(
    deep_branch_depth: NonNegativeSignedInt32,
    deep_branch_limit: NonNegativeSignedInt32,
) -> FetchRequest[TValue]

Specifies a limit on the number of results returned for each deep branch.

Notes

A deep branch has a root path that has a number of parts equal to the deep_branch_depth parameter. The deep_branch_limit specifies the maximum number of results for each deep branch.

This method is particularly useful for incrementally exploring a topic tree from the root, allowing a breadth-first search strategy.

For example, given a topic tree containing the topics with the following paths: z.

The z/5.

The fetch result does not indicate whether this option caused some results to be filtered from deep branches. It has no effect on the FetchResult.has_more result. If the result set contains deep_branch_limit results for a particular deep branch, some topics from that branch may have been filtered.

Parameters:

Name	Type	Description	Default
`deep_branch_depth`	`NonNegativeSignedInt32`	The number of parts in the root path of a branch for it to be considered deep.	required
`deep_branch_limit`	`NonNegativeSignedInt32`	The maximum number of results to return for each deep branch.	required

Returns:

Type	Description
`FetchRequest[TValue]`	The fetch request derived from this fetch request but restricting the number of results for deep sbranches.

with_unpublished_delayed_topics ¶

with_unpublished_delayed_topics() -> FetchRequest[TValue]

Include the details of reference topics that are not yet published.

Notes

[TopicViews][diffusion.features.topics.topic_views.TopicViews] that use the _delay by clause create reference topics in an unpublished state. The topics are published once the delay time has expired. A topic in the unpublished state prevents a lower priority topic view from creating a reference topic with the same path. _

A reference topic in the unpublished state which matches the query will only be included in the fetch results if the session has [PathPermission.READ_TOPIC][diffusion.PathPermission.READ_TOPIC] permission for the reference's source topic as well as READ_TOPIC permission for the reference topic. Requiring READ_TOPIC permission for the source topic ensures less privileged sessions cannot derive information from the existence of the reference topic before the delay time has expired. _

Added in version 6.11

Returns:

Type	Description
`FetchRequest[TValue]`	A fetch request derived from this fetch request, additionally specifying that unpublished reference topics should be included in the results

fetch `async` ¶

fetch(topics: StrictStr) -> FetchResult[TValue]

Sends a fetch request to the server.

Parameters:

Name	Type	Description	Default
`topics`	`StrictStr`	specifies a topic selector string which selects the topics to be fetched	required

Returns:

Type	Description
`FetchResult[TValue]`	the results of the fetch operation.

Raises:

Type	Description
`PermissionsException`	if the calling session does not have `SELECT_TOPIC` permission for the path prefix of the selector expression;
`SessionClosedException`	if the session is closed.

fetch_result ¶

TopicResult ¶

Bases: typing.Generic[TValue_Not_WildCard]

The result of a FetchRequest.fetch invocation for a single selected topic.

path `instance-attribute` ¶

path: str

The topic path.

specification `instance-attribute` ¶

specification: TopicSpecification

The topic specification.

Notes

If the request specified FetchRequest.with_properties, the result reflects the topic's specification and can be used to create an identical topic. If the request did not specify FetchRequest.with_properties, the specification's property map will be empty.

value `instance-attribute` ¶

value: typing.Optional[TValue_Not_WildCard]

The topic value or None if none available.

type `property` ¶

type: typing.Type[TValue_Not_WildCard]

Gets the topic type.

Returns:

Type	Description
`typing.Type[TValue_Not_WildCard]`	The topic type.

FetchResult ¶

Bases: typing.Generic[TValue]

The FetchResult[TValue] implementation.

results `instance-attribute` ¶

results: typing.Tuple[
    TopicResult[FetchQueryValueType_Not_WildCard], ...
]

The results

has_more `instance-attribute` ¶

has_more: bool

Whether there are more results

create `classmethod` ¶

create(
    tp_raw: typing.Type[TNewValue],
    result: FetchQueryResult,
    generic_converter: Optional[GenericConverter] = None,
) -> FetchResult[TNewValue]

Creates a FetchResult[TNewValue].

Parameters:

Name	Type	Description	Default
`tp_raw`	`typing.Type[TNewValue]`	The type of the result.	required
`result`	`FetchQueryResult`	The fetch query result.	required
`generic_converter`	`Optional[GenericConverter]`	Custom type converter.	`None`

Returns:

Type	Description
`FetchResult[TNewValue]`	The newly created fetch result.

read_as `classmethod` ¶

read_as(
    value_type: typing.Type[TNewValue_Not_WildCard],
    bytes_value: typing.Optional[bytes],
    data_type: typing.Type[
        FetchQueryValueType_Not_WildCard
    ],
    generic_converter: typing.Optional[
        GenericConverter
    ] = None,
) -> typing.Optional[TNewValue_Not_WildCard]

read_as(
    value_type: typing.Type[FetchQueryValueType_WildCards],
    bytes_value: typing.Optional[bytes],
    data_type: typing.Type[
        FetchQueryValueType_Not_WildCard
    ],
    generic_converter: typing.Optional[
        GenericConverter
    ] = None,
) -> typing.Optional[FetchQueryValueType_Not_WildCard]

read_as(
    value_type: typing.Type[
        typing.Union[
            TNewValue_Not_WildCard,
            FetchQueryValueType_WildCards,
        ]
    ],
    bytes_value: typing.Optional[bytes],
    data_type: typing.Type[
        FetchQueryValueType_Not_WildCard
    ],
    generic_converter: typing.Optional[
        GenericConverter
    ] = None,
) -> typing.Optional[FetchQueryValueType_Not_WildCard]

Parameters:

Name	Type	Default
`value_type`	`typing.Type[typing.Union[TNewValue_Not_WildCard, FetchQueryValueType_WildCards]]`	required
`bytes_value`	`typing.Optional[bytes]`	required
`data_type`	`typing.Type[FetchQueryValueType_Not_WildCard]`	required
`generic_converter`	`typing.Optional[GenericConverter]`	`None`

types ¶

TimeSeriesOfBytes `module-attribute` ¶

TimeSeriesOfBytes: TypeAlias = TimeSeriesEventDataType[
    Bytes
]

A timeseries with Bytes values

FetchQueryValueType_Diffusion `module-attribute` ¶

FetchQueryValueType_Diffusion = typing.Union[
    diffusion.datatypes.JSON,
    diffusion.datatypes.BINARY,
    diffusion.datatypes.DOUBLE,
    diffusion.datatypes.INT64,
    diffusion.datatypes.STRING,
    diffusion.datatypes.RECORD_V2,
    TimeSeriesOfBytes,
]

Concrete Fetch Query Value Types

FetchQueryValueType_Not_WildCard `module-attribute` ¶

FetchQueryValueType_Not_WildCard = typing.Union[
    FetchQueryValueType_Diffusion, IBytes
]

Fetch Query Value Type excluding wildcards

FetchQueryValueType_WildCards `module-attribute` ¶

FetchQueryValueType_WildCards = typing.Union[
    Object, IVoidFetch
]

Fetch Query Value Type wildcards

FetchQueryValueType `module-attribute` ¶

FetchQueryValueType = typing.Union[
    FetchQueryValueType_Not_WildCard,
    FetchQueryValueType_WildCards,
]

A Fetch Query Value type

FetchQueryValueType_Not_IVoidFetch `module-attribute` ¶

FetchQueryValueType_Not_IVoidFetch = typing.Union[
    FetchQueryValueType_Not_WildCard, Object
]

A Fetch Query Value type, excluding IVoidFetch

UnfrozenTopicTypeSet `module-attribute` ¶

UnfrozenTopicTypeSet = typing.Set[
    typing.Type[FetchQueryValueType_Not_WildCard]
]

A set of Fetch Query Value Types, not including wildcards

FrozenTopicTypeSet `module-attribute` ¶

FrozenTopicTypeSet = typing.FrozenSet[
    typing.Type[FetchQueryValueType_Not_WildCard]
]

A frozen set of Fetch Query Value Types, not including wildcards

TopicTypeSet `module-attribute` ¶

TopicTypeSet = typing.Union[
    FrozenTopicTypeSet, UnfrozenTopicTypeSet
]

Type alias of typing.Union[FrozenTopicTypeSet, FrozenTopicTypeSet]

TValue `module-attribute` ¶

TValue = typing.TypeVar("TValue", bound=FetchQueryValueType)

Value type of a fetch query

TNewValue `module-attribute` ¶

TNewValue = typing.TypeVar(
    "TNewValue", bound=FetchQueryValueType
)

New value type of a fetch query

TValue_Not_WildCard `module-attribute` ¶

TValue_Not_WildCard = typing.TypeVar(
    "TValue_Not_WildCard",
    bound=FetchQueryValueType_Not_WildCard,
    covariant=True,
)

A Fetch Query Value type, but not a wildcard

TNewValue_Not_WildCard `module-attribute` ¶

TNewValue_Not_WildCard = typing.TypeVar(
    "TNewValue_Not_WildCard",
    bound=FetchQueryValueType_Not_WildCard,
)

A new Fetch Query Value type, but not a wildcard

TNewValue_Not_IVoidFetch `module-attribute` ¶

TNewValue_Not_IVoidFetch = typing.TypeVar(
    "TNewValue_Not_IVoidFetch",
    bound=FetchQueryValueType_Not_IVoidFetch,
)

A new Fetch Query Value type, but not an IVoidFetch

TNewValue_Not_WildCard_Other `module-attribute` ¶

TNewValue_Not_WildCard_Other = typing.TypeVar(
    "TNewValue_Not_WildCard_Other",
    bound=FetchQueryValueType_Not_WildCard,
)

Another Fetch Query Value type, but not a wildcard

diffusion.features.topics.UnsubscribeReason ¶

Bases: IntEnum

The reason that an unsubscription occurred.

REQUESTED `class-attribute` `instance-attribute` ¶

REQUESTED = 0

Unsubscribed by the subscribing session.

CONTROL `class-attribute` `instance-attribute` ¶

CONTROL = 1

The unsubscription was requested either by another session or by the server.

REMOVAL `class-attribute` `instance-attribute` ¶

REMOVAL = 2

The unsubscription occurred because the topic was removed.

AUTHORIZATION `class-attribute` `instance-attribute` ¶

AUTHORIZATION = 3

The unsubscription occurred because the session is no longer authorized to access the topic.

UNKNOWN_UNSUBSCRIBE_REASON `class-attribute` `instance-attribute` ¶

UNKNOWN_UNSUBSCRIBE_REASON = 4

A reason that is unsupported by the session.

BACK_PRESSURE `class-attribute` `instance-attribute` ¶

BACK_PRESSURE = 5

The server has a significant backlog of messages for the session, and the topic specification has the conflation topic property set to "unsubscribe".

BRANCH_MAPPINGS `class-attribute` `instance-attribute` ¶

BRANCH_MAPPINGS = 6

The unsubscription occurred because branch mapping rules changed.

Topics¶

diffusion.features.topics.Topics ¶

topics property ¶

fetch_request ¶

add_topic async ¶

add_and_set_topic async ¶

set_topic async ¶

remove_topic async ¶

add_value_stream ¶

add_fallback_stream ¶

subscribe async ¶

unsubscribe async ¶

remove_stream async ¶

diffusion.features.topics.details.topic_specification ¶

T_other module-attribute ¶

WithProperties ¶

CompressionPolicy ¶

OFF class-attribute instance-attribute ¶

LOW class-attribute instance-attribute ¶

MEDIUM class-attribute instance-attribute ¶

HIGH class-attribute instance-attribute ¶

TopicDeliveryPriority ¶

LOW class-attribute instance-attribute ¶

DEFAULT class-attribute instance-attribute ¶

HIGH class-attribute instance-attribute ¶

ConflationPolicy ¶

OFF class-attribute instance-attribute ¶

CONFLATE class-attribute instance-attribute ¶

UNSUBSCRIBE class-attribute instance-attribute ¶

ALWAYS class-attribute instance-attribute ¶

TopicSpecification ¶

Topic Properties¶

topic_type property ¶

with_properties class-attribute ¶

PUBLISH_VALUES_ONLY class-attribute instance-attribute ¶

VALIDATE_VALUES class-attribute instance-attribute ¶

TIDY_ON_UNSUBSCRIBE class-attribute instance-attribute ¶

SCHEMA class-attribute instance-attribute ¶

DONT_RETAIN_VALUE class-attribute instance-attribute ¶

PERSISTENT class-attribute instance-attribute ¶

REMOVAL class-attribute instance-attribute ¶

CONFLATION class-attribute instance-attribute ¶

OWNER class-attribute instance-attribute ¶

COMPRESSION class-attribute instance-attribute ¶

PRIORITY class-attribute instance-attribute ¶

new_topic_specification classmethod ¶

diffusion.features.topics.streams ¶

LOG module-attribute ¶

SubHandlerValues ¶

unsubscribe instance-attribute ¶

subscribe instance-attribute ¶

update instance-attribute ¶

close instance-attribute ¶

ValueStreamHandler ¶

__init__ ¶

diffusion.features.topics.selectors ¶

Selector ¶

raw property ¶

match ¶

TopicSelectorError ¶

PathSelector ¶

raw property ¶

match ¶

PatternSelector ¶

FullPathPatternSelector ¶

match ¶

SplitPathPatternSelector ¶

match ¶

SelectorSet ¶

raw property ¶

match ¶

any_of classmethod ¶

get_selector ¶

diffusion.features.topics.update ¶

UpdateConstraint ¶

__and__ ¶

constraint_factory ¶

ConstraintFactory ¶

json_value property ¶

no_value property ¶

topics `property` ¶

add_topic `async` ¶

add_and_set_topic `async` ¶

set_topic `async` ¶

remove_topic `async` ¶

subscribe `async` ¶

unsubscribe `async` ¶

remove_stream `async` ¶

T_other `module-attribute` ¶

OFF `class-attribute` `instance-attribute` ¶

LOW `class-attribute` `instance-attribute` ¶

MEDIUM `class-attribute` `instance-attribute` ¶

HIGH `class-attribute` `instance-attribute` ¶

LOW `class-attribute` `instance-attribute` ¶

DEFAULT `class-attribute` `instance-attribute` ¶

HIGH `class-attribute` `instance-attribute` ¶

OFF `class-attribute` `instance-attribute` ¶

CONFLATE `class-attribute` `instance-attribute` ¶

UNSUBSCRIBE `class-attribute` `instance-attribute` ¶

ALWAYS `class-attribute` `instance-attribute` ¶

topic_type `property` ¶

with_properties `class-attribute` ¶

PUBLISH_VALUES_ONLY `class-attribute` `instance-attribute` ¶

VALIDATE_VALUES `class-attribute` `instance-attribute` ¶

TIDY_ON_UNSUBSCRIBE `class-attribute` `instance-attribute` ¶

SCHEMA `class-attribute` `instance-attribute` ¶

DONT_RETAIN_VALUE `class-attribute` `instance-attribute` ¶

PERSISTENT `class-attribute` `instance-attribute` ¶

REMOVAL `class-attribute` `instance-attribute` ¶

CONFLATION `class-attribute` `instance-attribute` ¶

OWNER `class-attribute` `instance-attribute` ¶

COMPRESSION `class-attribute` `instance-attribute` ¶

PRIORITY `class-attribute` `instance-attribute` ¶

new_topic_specification `classmethod` ¶

LOG `module-attribute` ¶

unsubscribe `instance-attribute` ¶

subscribe `instance-attribute` ¶

update `instance-attribute` ¶

close `instance-attribute` ¶

init ¶

raw `property` ¶

raw `property` ¶

raw `property` ¶

any_of `classmethod` ¶

and ¶

json_value `property` ¶

no_value `property` ¶

no_topic `property` ¶

PartialJSONType `module-attribute` ¶

encode `classmethod` ¶

decode `classmethod` ¶

path `class-attribute` `instance-attribute` ¶

includes_path `class-attribute` `instance-attribute` ¶

selector `class-attribute` `instance-attribute` ¶

range `class-attribute` `instance-attribute` ¶

topic_types `class-attribute` `instance-attribute` ¶

with_values `class-attribute` `instance-attribute` ¶

with_properties `class-attribute` `instance-attribute` ¶

limit `class-attribute` `instance-attribute` ¶

maximum_result_size `class-attribute` `instance-attribute` ¶

deep_branch_depth `class-attribute` `instance-attribute` ¶

deep_branch_limit `class-attribute` `instance-attribute` ¶

with_unpublished_delayed_topics `class-attribute` `instance-attribute` ¶

FetchContext `dataclass` ¶

session `instance-attribute` ¶

maximum_message_size `instance-attribute` ¶

fetch `async` ¶

path `instance-attribute` ¶

specification `instance-attribute` ¶

value `instance-attribute` ¶

type `property` ¶

results `instance-attribute` ¶

has_more `instance-attribute` ¶

create `classmethod` ¶

read_as `classmethod` ¶

TimeSeriesOfBytes `module-attribute` ¶