Skip to content

Restructure code to be more Pythonic and adhere to PEP8 style #340

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Mar 27, 2018
Merged

Restructure code to be more Pythonic and adhere to PEP8 style #340

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b5f51e4
Restructure code to be more Pythonic and adhere to PEP8 style
Mar 23, 2018
cf40419
Move try/except closer to source
Mar 23, 2018
9f0c0d2
Condition on msg is None instead
Mar 26, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
74 changes: 53 additions & 21 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,9 +39,12 @@ Usage
```python
from confluent_kafka import Producer


p = Producer({'bootstrap.servers': 'mybroker,mybroker2'})

for data in some_data_source:
p.produce('mytopic', data.encode('utf-8'))

p.flush()
```

Expand All @@ -51,17 +54,29 @@ p.flush()
```python
from confluent_kafka import Consumer, KafkaError

c = Consumer({'bootstrap.servers': 'mybroker', 'group.id': 'mygroup',
'default.topic.config': {'auto.offset.reset': 'smallest'}})

c = Consumer({
'bootstrap.servers': 'mybroker',
'group.id': 'mygroup',
'default.topic.config': {
'auto.offset.reset': 'smallest'
}
})

c.subscribe(['mytopic'])
running = True
while running:

while True:
msg = c.poll()
if not msg.error():
print('Received message: %s' % msg.value().decode('utf-8'))
elif msg.error().code() != KafkaError._PARTITION_EOF:
print(msg.error())
running = False

if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
else:
print(msg.error())
break

print('Received message: {}'.format(msg.value().decode('utf-8')))

c.close()
```

Expand All @@ -71,12 +86,17 @@ c.close()
from confluent_kafka import avro
from confluent_kafka.avro import AvroProducer


value_schema = avro.load('ValueSchema.avsc')
key_schema = avro.load('KeySchema.avsc')
value = {"name": "Value"}
key = {"name": "Key"}

avroProducer = AvroProducer({'bootstrap.servers': 'mybroker,mybroker2', 'schema.registry.url': 'http://schem_registry_host:port'}, default_key_schema=key_schema, default_value_schema=value_schema)
avroProducer = AvroProducer({
'bootstrap.servers': 'mybroker,mybroker2',
'schema.registry.url': 'http://schem_registry_host:port'
}, default_key_schema=key_schema, default_value_schema=value_schema)

avroProducer.produce(topic='my_topic', value=value, key=key)
avroProducer.flush()
```
Expand All @@ -88,21 +108,33 @@ from confluent_kafka import KafkaError
from confluent_kafka.avro import AvroConsumer
from confluent_kafka.avro.serializer import SerializerError

c = AvroConsumer({'bootstrap.servers': 'mybroker,mybroker2', 'group.id': 'groupid', 'schema.registry.url': 'http://127.0.0.1:8081'})

c = AvroConsumer({
'bootstrap.servers': 'mybroker,mybroker2',
'group.id': 'groupid',
'schema.registry.url': 'http://127.0.0.1:8081'})

c.subscribe(['my_topic'])
running = True
while running:

while True:
try:
msg = c.poll(10)
if msg:
if not msg.error():
print(msg.value())
elif msg.error().code() != KafkaError._PARTITION_EOF:
print(msg.error())
running = False

except SerializerError as e:
print("Message deserialization failed for %s: %s" % (msg, e))
running = False
print("Message deserialization failed for {}: {}".format(msg, e))
break

if msg is None:
continue

if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
else:
print(msg.error())
break

print(msg.value())

c.close()
```
Expand Down