feat: improve topic creation api

[fraidev]

Simple (1 partition and 1 replica):

fluvio_admin = FluvioAdmin.connect()
fluvio_admin.create_topic("topic_name")

or

fluvio_admin = FluvioAdmin.connect()
topic_spec = TopicSpec.new()
fluvio_admin.create_topic("topic_name", topic_spec)

Advanced:

fluvio_admin = FluvioAdmin.connect()
topic_spec = (
    TopicSpec.create()
    .with_partitions(3)
    .with_replications(3)
    .with_max_partition_size("1Gb")
    .with_retention_time(3600)
    .with_segment_size("10M")
    .as_system_topic()
    .build()
)
fluvio_admin.create_topic("topic_name", topic_spec)

I am using NewTopic only because kafka-python also uses a class called NewTopic

[digikata]

Instead of using Kafka terms, lets stay with the terminology in the Rust crate, TopicSpec. PyO3 lets us provide good inline python docs, but if those are insufficient, people (including ourselves) often end up referring to the Rust crate code. So staying consistent with that is a principle.

[sehz]

Instead of creating python enum, should use derive macro like here: https://pyo3.rs/v0.23.2/conversions/traits.html

[fraidev]

This new enum is only for the builder of TopicSpec, if we consider the solution without the build it's not even needed.

[fraidev]

Maybe the Compression Enum is worth, thought

[fraidev]

Instead of using Kafka terms, lets stay with the terminology in the Rust crate, TopicSpec. PyO3 lets us provide good inline python docs, but if those are insufficient, people (including ourselves) often end up referring to the Rust crate code. So staying consistent with that is a principle.

Sure, but should I merge them or rename the actual TopicSpec?

Because the actual TopicSpec is a wrapper of the inner TopicSpec from Rust.

The NewTopic that I created is the builder of TopicSpec. Allowing specifying the number of the partitions and replications with these methods, instead just call new_computed method.

Merge them would be something like this, what do you think?

fluvio_admin = FluvioAdmin.connect()
topic_spec = (
    TopicSpec.new_computed(partitions=3, replications=3, ignore=True)
    .with_max_partition_size("1Gb")
    .with_retention_time(3600)
    .with_segment_size("10M")
    .as_system_topic()
)
fluvio_admin.create_topic(self.topic, topic_spec)

Another alternative is rename NewTopic to TopicSpecBuilder or rename the NewTopic to TopicSpec and TopicSpec to something else.

fluvio_admin = FluvioAdmin.connect()
topic_spec = (
    TopicSpecBuilder.create()
    .with_partitions(3)
    .with_replications(3)
    .with_max_partition_size("1Gb")
    .with_retention_time(3600)
    .with_segment_size("10M")
    .as_system_topic()
    .build()
)
fluvio_admin.create_topic(self.topic, topic_spec)

[digikata]

The first one is good w/ TopicSpec. If you can do it with the direct annotations in lib.rs PyO3 or appropriate module use that. There are a lot of instances in the fluvio python client that rename some FluvioStruct to _FluvioStruct internally, so a pythonic FluvioStruct can be supplied to the interpreter too, but it's extra layers that should have some interface value if thats done. (there are legacy portions of the code that does it unecessarily)

[digikata]

Oh we shoudl also remove the required partitions, replications, and bool arguments and just support TopicSpec.new() with defaults that can be changed w/ the builder style calls as you did with the other parameters

[fraidev]

Oh we shoudl also remove the required partitions, replications, and bool arguments and just support TopicSpec.new() with defaults that can be changed w/ the builder style calls as you did with the other parameters

I think that would need a larger refactor on fluvio side. IIRC, we don't have the feature of adding replicas to an already existent Topic.

In the builder it was easier because it's calling new only after the build function.

wdyt?

[digikata]

This is only more featured topic create for the python client, if the topic already exists the create should return an error.

[digikata]

This should probably return an error if not recognized, or it could accept a compression type enum

[fraidev]

I changed to return result, and an error for this case.
The support for enum that are not tag unions does not seem great in pyo3, it's also not automatically typed, so no benefits at all.