neuroglia-io / python-framework Goto Github PK

A MongoQuery may return an inaccurate empty list after multiple requests with different payloads.

For example, referring to the following query, sending a first query Q1 with a "report_type:A" parameter will return a list of - for example - 3 ReportDto's. A second query Q2 with a "report_type:B" parameter will return an empty list, which is correct in the test.
A third query Q3 with a "report_type:A" parameter (so the same query as Q1!) will now return an empty list - which is wrong as it should still be 3...

class ReportsByTypeQuery(Query[OperationResult[ReportDto]]):
    """Represents the query used to get an account by its owner id"""

    report_type: ReportType

    def __init__(self, report_type: ReportType):
        self.report_type = report_type


class ReportsByTypeQueryHandler(
    QueryHandler[ReportsByTypeQuery, OperationResult[ReportDto]]
):
    """Represents the service used to handle ReportsByTypeQuery instances"""

    def __init__(self, repository: QueryableRepository[ReportDto, str]):
        self.repository = repository

    repository: QueryableRepository[ReportDto, str]

    async def handle_async(
        self, query: ReportsByTypeQuery
    ) -> OperationResult[ReportDto]:
        # q = await self.repository.query_async()
        # result = q.where(lambda r: r.type == query.report_type.value).to_list()
        # return self.ok(result)
        return self.ok((await self.repository.query_async()).where(lambda r: r.type == query.report_type.value).to_list())

I'm using the neuroglia.data.infrastructure.mongo.mongo_repository module... Thanks!

Would be great to carry the CloudEvent source attribute along with the type so that event consumers can use the information if needed.

Currently, the CloudEventIngestor maps a CloudEvent to an IntegrationEvent based on the type attribute and instantiates the event based on its data payload only, which makes perfect sense. However, if the event payload does not include anything specific about the source, then the IntegrationEvent consumer has no way to retrieve the information (as it's lost in the enveloppe that was just removed).

It'd be great to carry that info in the IntegrationEvent along with the payload.

It'd be super awesome if the framework was available as a pip package.

It doesnt matter as much if it's hosted somewhere else than the actual PyPi registry...

BTW, i do have a branch with this done but cant push it here.

It would be even more awesome if the framework supported persistent subscriptions :)