airtai / fastkafka Goto Github PK

Looks good but giant freeform dictionaries for config data always give me pause for thought. You're already using pydantic, might be worth using some pydantic classes for safer config validation.

https://www.reddit.com/r/Python/comments/11paz9u/comment/jbydm6j/?utm_source=share&utm_medium=web2x&context=3

I really like the idea of this, as the biggest gripe I have with most pub/sub solutions is all of the tedious boiler plate code needed to correctly subscribe and publish and manage message leases etc. While you often just want to grab a message, do some processing and put it on a different queue.

One of the most obvious improvements would be supporting more pubsub backends (thinking about AWS SQS, google cloud pubsub, RabbitMQ, ...)

https://www.reddit.com/r/Python/comments/11paz9u/comment/jbxf1v8/?utm_source=share&utm_medium=web2x&context=3

e.g.

fastkafka aws deploy kafka_service:kafka_app --working_dir=. --param1=value1 --param2=value2
--name=production-kafka-service
--requirements-file-name=requirements.txt
--instance-type=m5.xlarge

• some kind of docker based volume mapping etc.

(potentially some other params that are supported by AWS) there will be some other things to look for

fastkafka aws ls

fastkafka aws stop --name=production-kafka-service

• Implement @process decorator
• Write and execute tests for it
• Create and write a new guide for the decorator
• transactional semantics?

Previous syntax for admonitions was:

!!! info \"Title\"

    Text

New syntax, after moving to Docusaurus is:

:::info Title

Text

:::

The issue is, that the documentation is being first parsed by mkdocs, and ::: is a keyword: https://mkdocstrings.github.io/usage/
So we cant just replace !!! with :::.

See: https://fastkafka.airt.ai/docs/api/fastkafka/testing/Tester

Remarks:
async functions produce faster than sync functions, add recommendations to use them whenever possible

I'd like to use something like this:

def make_func_push(func: Callable) -> Callable:
    """Decores func or method, making it push metrics into Prometeus."""
    return prometheus_pusher(metrics=[metrics.metric_factory.get_metrics()], job=pod_name)(metrics.calculate(func))


class PushingProducer(AIOKafkaProducer):
    """Pushing to Prometeus Kafka producer."""

    send = make_func_push(AIOKafkaProducer.send)
    send_batch = make_func_push(AIOKafkaProducer.send_batch)

as producer for whole application by default or with chosen producers.

For now, monkeypatch is an option, but I'd thank much for more flexible solution.

https://fastkafka.airt.ai/docs/next/api/fastkafka/#fastkafka._application.app.FastKafka.consumes

@pytest.fixture
async def app(fastkafka_tester: Tester):
    app = prepare_fastapi_app()
    async with fastkafka_tester.fastapi_lifespan(settings.KAFKA.NAME)(app):
        yield app

If I'm using this fixture to test FastAPI application on sequence, having producer's call, I'm receiving

>       fut = await producer.send(
            topic, encoder_fn(wrapped_val.message), key=wrapped_val.key
        )
E       AttributeError: 'NoneType' object has no attribute 'send'

Because producers were not instantiated properly.
Am I doing something wrong?

Also this works:

@pytest.fixture
async def app(fastkafka_app: FastKafka):
    app = prepare_fastapi_app()
    async with fastkafka_app.fastapi_lifespan(settings.KAFKA.NAME)(app):
        yield app

Async docs generation fails when Tuples are nested in examples
Async docs generation fails when Enum is defined as one of the values

Needs more exploration and reproducible example

We should link the Using FastAPI to run FastKafka guide in the README so that this functionality is more visible

• New account(email, password) for confluent CLI
• Add payment -

    Also, how can I skip a commit if a critical error has occurred? I would like this message to return in the topic and be read, for example, after a restart. It would be cool if the problem of message loss was solved.

Originally posted by @fangorntreabeard in https://github.com/airtai/fastkafka/discussions/138#discussioncomment-5372172

As a small possibly suggestion, perhaps to make possible fallback of being able to work on Sync stuff with process workers?

Recently tried FastAPI and was not fascinated that it works in anyio async engine which severely crippled celery working in a normal way. (Broken task discovery, not working global initialised stuff) Too many async engines around with no uniform compatibility to them.

Having option to run as sync / processes can be good idea for stability of usage for prod

https://discordapp.com/channels/267624335836053506/366673702533988363/1086276920662040637

You should tweak the presentation to introduce a concept on a simple example and then show how an actual project would use it. The actual project should be something concrete (separate repo), using only the standard Python toolkit (no notebook and nbdev). The Iris model is fine, but maybe we can find something more exciting to use.

Does that assume that Kafka messages are in JSON format. What if we wanna use protobuf for example?

https://www.reddit.com/r/FastAPI/comments/11oq09r/comment/jbx2yrq/?utm_source=share&utm_medium=web2x&context=3

• dataclasses (should be easy to add as pydantic can work with them out of the box)
• msgspec (high performance serialization & validation library)

• support TLS in constructor and implementation
• write a guide for it

https://discord.com/channels/1085457301214855171/1085457302280228898/1118857149481750538

Check 001_ConsumerLoop _decode_msgs_and_send function.

NOTE: If type bytes, the message should not be decoded but sent to consumer callback as is so that the user can then manually decode the message inside consumer callback