Various functionality questions

[reritom]

I like the look of this project and was looking at integrating it into a project of mine but have various questions. Some of these points might already apply and have been missed by me.

1 - Having used FastAPI and seen the various mentions on here, it would be nice to see some of their other documentation features:

• Being able to specfy "include_in_schema" in the publisher/subscriber - this is in fastapi. I personally use it to be able to build both an internal-facing and client-facing documentation for my service. Generally in the form of @router.post("/..", include_in_schema=settings.BUILD_INTERNAL_DOCS)
• Being able to specify the operation id in the publisher/subscriber - easier to reference flows in the documentation

2 - I dont see anything in the examples about direct publishing where the publish is part of the schema. From my fastapi application I would like to trigger direct publishes that appear in the docs. Perhaps something like the below example (maybe this is already the case, but I didn't see it in then docs and the they explicitly mentioned somewhere that direct publishes aren't documented, which makes sense when doing something like await broker.publish(...)):

@broker.publisher("topic")
def publish_user(user: UserModel) -> UserModel:
    return user

publish_user(user=...)

3 - Having some sort of dynamic documented publishes would be nice. Similar to the request for #764 but for providers with dynamic routing keys or topics.

Imagine receiving an entity, and then forwarding it to another specific topic based on the content of the payload.
Below is just an example of what I would like to achieve.

@broker.subscriber("user-created")
@broker.publisher("organisation-{organisationId}-new-user", routing_key="something-{dynamic}")
def forward_new_user_to_tenent_specific_queues(user: UserModel) -> UserModel:
    # Somehow set the topic/routing-key in various ways
    return user

Perhaps either:

• Allow the string interpolation using the single returned object, like @broker.publisher("organisation-{entity.organisationId}-new-user", routing_key="something-{entity.something}"), but this would be limiting values to stuff published in the payload.
• Something with context?
• I would go for something similar to manually created a Response in fastapi, so:

@broker.subscriber("user-created")
@broker.publisher("organisation-{organisationId}-new-user", routing_key="something-{dynamic}")
def forward_new_user_to_tenent_specific_queues(user: UserModel) -> EventToPublish[UserModel]:
    return EventToPublish(
        payload=user,
        topic="organisation-{organisationId}-new-user".format(...)
        routing_key="something-{dynamic}".format(...)
    )

Or the above case but perhaps add the routing_key and topics as a sort of dependency

@broker.subscriber("user-created")
@broker.publisher("organisation-{organisationId}-new-user", routing_key="something-{dynamic}")
def forward_new_user_to_tenent_specific_queues(user: UserModel, routing_key=..., topic =... ) -> EventToPublish[UserModel]:
    # The routing_key/topic come from some sort of Depends()
    # or by asking the `broker.publisher(...)` call to provide them to the publisher func.
    return EventToPublish(
        payload=user,
        topic=topic.format(...)
        routing_key=routing_key.format(...),
        headers=... #?
    )

4 - It wasn't immediately clear to me in the documentation to what extent certain objects perform logic, or whether they are just for generating documentation (queues[auto-delete,bind-arguments], exchanges[type], etc)

[Lancetnik]

First of all, thank you for such a close study of the documentation
Let's deal with everything in order

1. for now, you can't disable specific subscriber/router representation (thx for the idea), but you can specify their name by title="name" argument

2. For now you can create reusable and documentated publisher like

publisher = broker.publisher("response")
# and call it later
await publisher.publish("message")

3. Thanks for the idea one more time
   For now I image this behavior like extra publisher option or smth close to it

publisher = broker.publisher("response.%s")
# and call it later
await publisher.publish("message", extra=[user_id])

4. All these arguments are meaningful for the broker and using to specify your subscriber/object options
   As an example, take a look at RMQ header exchange using
   Also, some of these arguments are using for documentation generation as well, but it is not the major their function

[davorrunje]

2 - I dont see anything in the examples about direct publishing where the publish is part of the schema. From my fastapi application I would like to trigger direct publishes that appear in the docs. Perhaps something like the below example (maybe this is already the case, but I didn't see it in then docs and the they explicitly mentioned somewhere that direct publishes aren't documented, which makes sense when doing something like await broker.publish(...)):

@broker.publisher("topic")
def publish_user(user: UserModel) -> UserModel:
    return user

publish_user(user=...)

Please correct me if I am wrong, but you would like to have the schema of the publisher documented in the generated documentation. Something like this would then do the trick:

publisher = broker.publisher("response", schema=UserModel)
# and call it later
await publisher.publish("message")

[reritom]

You're right. I was fixating a bit on having a unified approach of using decorated functions in both cases.

Something like having my FastAPI routers.py where each interface looks the same using decorators. Then I could have a publishers.py/subsribers.py/events.py where all the interfaces are defined in the same way.

# routers.py
# The fastapi router definition file

router = ...

@router.post(...)
def create_user(user: User):
    # Persist it
    ...
    # Share it
    publish_user(user)
    return User

# events.py
# Where I define all of the pub/subs for this subapp
# Note that all of the definitions use the same decorated approach

import broker # from somewhere

# Some will be directly callable to publish
@broker.publisher(...)
def publish_user(user: User):
    ...

# Some will only listen
@broker.subscriber(...)
def do_something_with_user(...):
    ...

# Some will sub and pub (this seems like it would be difficult to call directly in the case where the handler uses the context or wants other subscriber specific things)
@broker.publisher(...)
@broker.subscriber(...)
def listen_to_something_else(...):
    ...

But yeah for general case of direct publishes that are documented, what you suggested satisfies that. So I assume my above example would be more like:

# events.py

broker = ...
new_user_publisher = broker.publisher(...)

def publish_user(user: User):
    new_user_publisher.publish(...)

I can imagine this approach could be a bit verbose if we are working with a lot of different publishers, where each one needs defining.

[Lancetnik]

@reritom it is not verbose if you don't create funtions like this:

async def publihs_user(user):
    await producer.publish(user)

Just use producer.publish(...) right in place you need: it is a much explicit way than creating functions like

@broker.publisher(...)
def publish_user(user: User):
    ...

# And call them later
publish_user(user)

Also, you can use the producer as a wrapper (if function is already wrapped by subscriber

@producer
@broker.subscriber("in")
async def handler(msg: str) -> str: ...

[Lancetnik]

Btw, you need no to create producers only with broker

producer = router.publisher("out")

Works fine too