Skip to content

Commit

Permalink
Merge pull request #24 from fabregas/documentation
Browse files Browse the repository at this point in the history
Documentation
  • Loading branch information
tvoinarovskyi committed Apr 10, 2016
2 parents cbbcde1 + 3b1b4a4 commit 2d95376
Show file tree
Hide file tree
Showing 14 changed files with 1,049 additions and 19 deletions.
10 changes: 5 additions & 5 deletions LICENSE
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
Apache License

Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/

Expand Down Expand Up @@ -178,15 +179,15 @@ Apache License
APPENDIX: How to apply the Apache License to your work.

To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "{}"
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.

Copyright {yyyy} {name of copyright owner}
Copyright 2016 Taras Voinarovskyi

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
Expand All @@ -198,5 +199,4 @@ Apache License
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

limitations under the License.
1 change: 1 addition & 0 deletions Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ clean:
rm -rf build
rm -rf cover
rm -rf dist
rm -rf docs/_build

doc:
make -C docs html
Expand Down
10 changes: 6 additions & 4 deletions README.rst
Original file line number Diff line number Diff line change
Expand Up @@ -26,10 +26,12 @@ Example of AIOKafkaProducer usage:
def produce(loop):
producer = AIOKafkaProducer(loop=loop, bootstrap_servers='localhost:1234')
yield from producer.start()
future1 = producer.send('foobar', b'some_message_bytes')
future2 = producer.send('foobar', key=b'foo', value=b'bar')
future3 = producer.send('foobar', b'message for partition 1', partition=1)
yield from asyncio.wait([future1, future2, future3], loop=loop)
future = yield from producer.send('foobar', b'some_message_bytes')
# waiting send result
resp = yield from future
print("Message was produced to partition %i with offset %i"%(resp.partition, resp.offset))
resp = yield from producer.send_and_wait('foobar', key=b'foo', value=b'bar')
resp = yield from producer.send_and_wait('foobar', b'message for partition 1', partition=1)
yield from producer.stop()
loop = asyncio.get_event_loop()
Expand Down
62 changes: 52 additions & 10 deletions aiokafka/consumer.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,8 @@ class AIOKafkaConsumer(object):
The consumer will transparently handle the failure of servers in the Kafka
cluster, and adapt as topic-partitions are created or migrate between
brokers. It also interacts with the assigned kafka Group Coordinator node
to allow multiple consumers to load balance consumption of topics (requires
kafka >= 0.9.0.0).
to allow multiple consumers to load balance consumption of topics (feature
of kafka >= 0.9.0.0).
Arguments:
*topics (str): optional list of topics to subscribe to. If not set,
Expand Down Expand Up @@ -109,6 +109,12 @@ class AIOKafkaConsumer(object):
AIOKafkaConsumer supports Kafka API versions >=0.9 only.
If set to 'auto', will attempt to infer the broker version by
probing various APIs. Default: auto
Note:
Many configuration parameters are taken from Java Client:
https://kafka.apache.org/documentation.html#newconsumerconfigs
"""
def __init__(self, *topics, loop,
bootstrap_servers='localhost',
Expand Down Expand Up @@ -277,8 +283,9 @@ def commit(self, offsets=None):
This commits offsets only to Kafka.
The offsets committed using this API will be used on the first fetch
after every rebalance and also on startup. As such, if you needto store
offsets in anything other than Kafka, this API should not be used.
after every rebalance and also on startup. As such, if you need to
store offsets in anything other than Kafka, this API should not be
used.
Blocks until either the commit succeeds or an unrecoverable error is
encountered (in which case it is thrown to the caller).
Expand Down Expand Up @@ -421,8 +428,9 @@ def seek(self, partition, offset):
@asyncio.coroutine
def seek_to_committed(self, *partitions):
"""Seek to the committed offset for partitions
Arguments:
*partitions: optionally provide specific TopicPartitions,
partitions: optionally provide specific TopicPartitions,
otherwise default to all assigned partitions
Raises:
Expand Down Expand Up @@ -545,16 +553,36 @@ def _on_change_subscription(self):
def getone(self, *partitions):
"""
Get one message from Kafka
If no new messages occured, this method will wait it
If no new messages prefetched, this method will wait for it
Arguments:
partitions (List[TopicPartition]): The partitions that need
fetching message. If no one partition specified then all
subscribed partitions will be used
partitions (List[TopicPartition]): Optional list of partitions to
return from. If no partitions specified then returned message
will be from any partition, which consumer is subscribed to.
Returns:
instance of collections.namedtuple("ConsumerRecord",
ConsumerRecord
Will return instance of
.. code:: python
collections.namedtuple(
"ConsumerRecord",
["topic", "partition", "offset", "key", "value"])
Example usage:
.. code:: python
while True:
message = yield from consumer.getone()
topic = message.topic
partition = message.partition
# Process message
print(message.offset, message.key, message.value)
"""
assert all(map(lambda k: isinstance(k, TopicPartition), partitions))

Expand All @@ -580,6 +608,20 @@ def getmany(self, *partitions, timeout_ms=0):
Returns:
dict: topic to list of records since the last fetch for the
subscribed list of topics and partitions
Example usage:
.. code:: python
data = yield from consumer.getmany()
for tp, messages in data.items():
topic = tp.topic
partition = tp.partition
for message in messages:
# Process message
print(message.offset, message.key, message.value)
"""
assert all(map(lambda k: isinstance(k, TopicPartition), partitions))

Expand Down
177 changes: 177 additions & 0 deletions docs/Makefile
Original file line number Diff line number Diff line change
@@ -0,0 +1,177 @@
# Makefile for Sphinx documentation
#

# You can set these variables from the command line.
SPHINXOPTS =
SPHINXBUILD = sphinx-build
PAPER =
BUILDDIR = _build

# User-friendly check for sphinx-build
ifeq ($(shell which $(SPHINXBUILD) >/dev/null 2>&1; echo $$?), 1)
$(error The '$(SPHINXBUILD)' command was not found. Make sure you have Sphinx installed, then set the SPHINXBUILD environment variable to point to the full path of the '$(SPHINXBUILD)' executable. Alternatively you can add the directory with the executable to your PATH. If you don't have Sphinx installed, grab it from http://sphinx-doc.org/)
endif

# Internal variables.
PAPEROPT_a4 = -D latex_paper_size=a4
PAPEROPT_letter = -D latex_paper_size=letter
ALLSPHINXOPTS = -d $(BUILDDIR)/doctrees $(PAPEROPT_$(PAPER)) $(SPHINXOPTS) .
# the i18n builder cannot share the environment and doctrees with the others
I18NSPHINXOPTS = $(PAPEROPT_$(PAPER)) $(SPHINXOPTS) .

.PHONY: help clean html dirhtml singlehtml pickle json htmlhelp qthelp devhelp epub latex latexpdf text man changes linkcheck doctest gettext

help:
@echo "Please use \`make <target>' where <target> is one of"
@echo " html to make standalone HTML files"
@echo " dirhtml to make HTML files named index.html in directories"
@echo " singlehtml to make a single large HTML file"
@echo " pickle to make pickle files"
@echo " json to make JSON files"
@echo " htmlhelp to make HTML files and a HTML help project"
@echo " qthelp to make HTML files and a qthelp project"
@echo " devhelp to make HTML files and a Devhelp project"
@echo " epub to make an epub"
@echo " latex to make LaTeX files, you can set PAPER=a4 or PAPER=letter"
@echo " latexpdf to make LaTeX files and run them through pdflatex"
@echo " latexpdfja to make LaTeX files and run them through platex/dvipdfmx"
@echo " text to make text files"
@echo " man to make manual pages"
@echo " texinfo to make Texinfo files"
@echo " info to make Texinfo files and run them through makeinfo"
@echo " gettext to make PO message catalogs"
@echo " changes to make an overview of all changed/added/deprecated items"
@echo " xml to make Docutils-native XML files"
@echo " pseudoxml to make pseudoxml-XML files for display purposes"
@echo " linkcheck to check all external links for integrity"
@echo " doctest to run all doctests embedded in the documentation (if enabled)"

clean:
rm -rf $(BUILDDIR)/*

html:
$(SPHINXBUILD) -b html $(ALLSPHINXOPTS) $(BUILDDIR)/html
@echo
@echo "Build finished. The HTML pages are in $(BUILDDIR)/html."

dirhtml:
$(SPHINXBUILD) -b dirhtml $(ALLSPHINXOPTS) $(BUILDDIR)/dirhtml
@echo
@echo "Build finished. The HTML pages are in $(BUILDDIR)/dirhtml."

singlehtml:
$(SPHINXBUILD) -b singlehtml $(ALLSPHINXOPTS) $(BUILDDIR)/singlehtml
@echo
@echo "Build finished. The HTML page is in $(BUILDDIR)/singlehtml."

pickle:
$(SPHINXBUILD) -b pickle $(ALLSPHINXOPTS) $(BUILDDIR)/pickle
@echo
@echo "Build finished; now you can process the pickle files."

json:
$(SPHINXBUILD) -b json $(ALLSPHINXOPTS) $(BUILDDIR)/json
@echo
@echo "Build finished; now you can process the JSON files."

htmlhelp:
$(SPHINXBUILD) -b htmlhelp $(ALLSPHINXOPTS) $(BUILDDIR)/htmlhelp
@echo
@echo "Build finished; now you can run HTML Help Workshop with the" \
".hhp project file in $(BUILDDIR)/htmlhelp."

qthelp:
$(SPHINXBUILD) -b qthelp $(ALLSPHINXOPTS) $(BUILDDIR)/qthelp
@echo
@echo "Build finished; now you can run "qcollectiongenerator" with the" \
".qhcp project file in $(BUILDDIR)/qthelp, like this:"
@echo "# qcollectiongenerator $(BUILDDIR)/qthelp/aiokafka.qhcp"
@echo "To view the help file:"
@echo "# assistant -collectionFile $(BUILDDIR)/qthelp/aiokafka.qhc"

devhelp:
$(SPHINXBUILD) -b devhelp $(ALLSPHINXOPTS) $(BUILDDIR)/devhelp
@echo
@echo "Build finished."
@echo "To view the help file:"
@echo "# mkdir -p $$HOME/.local/share/devhelp/aiokafka"
@echo "# ln -s $(BUILDDIR)/devhelp $$HOME/.local/share/devhelp/aiokafka"
@echo "# devhelp"

epub:
$(SPHINXBUILD) -b epub $(ALLSPHINXOPTS) $(BUILDDIR)/epub
@echo
@echo "Build finished. The epub file is in $(BUILDDIR)/epub."

latex:
$(SPHINXBUILD) -b latex $(ALLSPHINXOPTS) $(BUILDDIR)/latex
@echo
@echo "Build finished; the LaTeX files are in $(BUILDDIR)/latex."
@echo "Run \`make' in that directory to run these through (pdf)latex" \
"(use \`make latexpdf' here to do that automatically)."

latexpdf:
$(SPHINXBUILD) -b latex $(ALLSPHINXOPTS) $(BUILDDIR)/latex
@echo "Running LaTeX files through pdflatex..."
$(MAKE) -C $(BUILDDIR)/latex all-pdf
@echo "pdflatex finished; the PDF files are in $(BUILDDIR)/latex."

latexpdfja:
$(SPHINXBUILD) -b latex $(ALLSPHINXOPTS) $(BUILDDIR)/latex
@echo "Running LaTeX files through platex and dvipdfmx..."
$(MAKE) -C $(BUILDDIR)/latex all-pdf-ja
@echo "pdflatex finished; the PDF files are in $(BUILDDIR)/latex."

text:
$(SPHINXBUILD) -b text $(ALLSPHINXOPTS) $(BUILDDIR)/text
@echo
@echo "Build finished. The text files are in $(BUILDDIR)/text."

man:
$(SPHINXBUILD) -b man $(ALLSPHINXOPTS) $(BUILDDIR)/man
@echo
@echo "Build finished. The manual pages are in $(BUILDDIR)/man."

texinfo:
$(SPHINXBUILD) -b texinfo $(ALLSPHINXOPTS) $(BUILDDIR)/texinfo
@echo
@echo "Build finished. The Texinfo files are in $(BUILDDIR)/texinfo."
@echo "Run \`make' in that directory to run these through makeinfo" \
"(use \`make info' here to do that automatically)."

info:
$(SPHINXBUILD) -b texinfo $(ALLSPHINXOPTS) $(BUILDDIR)/texinfo
@echo "Running Texinfo files through makeinfo..."
make -C $(BUILDDIR)/texinfo info
@echo "makeinfo finished; the Info files are in $(BUILDDIR)/texinfo."

gettext:
$(SPHINXBUILD) -b gettext $(I18NSPHINXOPTS) $(BUILDDIR)/locale
@echo
@echo "Build finished. The message catalogs are in $(BUILDDIR)/locale."

changes:
$(SPHINXBUILD) -b changes $(ALLSPHINXOPTS) $(BUILDDIR)/changes
@echo
@echo "The overview file is in $(BUILDDIR)/changes."

linkcheck:
$(SPHINXBUILD) -b linkcheck $(ALLSPHINXOPTS) $(BUILDDIR)/linkcheck
@echo
@echo "Link check complete; look for any errors in the above output " \
"or in $(BUILDDIR)/linkcheck/output.txt."

doctest:
$(SPHINXBUILD) -b doctest $(ALLSPHINXOPTS) $(BUILDDIR)/doctest
@echo "Testing of doctests in the sources finished, look at the " \
"results in $(BUILDDIR)/doctest/output.txt."

xml:
$(SPHINXBUILD) -b xml $(ALLSPHINXOPTS) $(BUILDDIR)/xml
@echo
@echo "Build finished. The XML files are in $(BUILDDIR)/xml."

pseudoxml:
$(SPHINXBUILD) -b pseudoxml $(ALLSPHINXOPTS) $(BUILDDIR)/pseudoxml
@echo
@echo "Build finished. The pseudo-XML files are in $(BUILDDIR)/pseudoxml."
37 changes: 37 additions & 0 deletions docs/api.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@

API Documentation
=================

AIOKafkaProducer class
----------------------

.. autoclass:: aiokafka.AIOKafkaProducer
:members:


AIOKafkaConsumer class
----------------------

.. autoclass:: aiokafka.AIOKafkaConsumer
:members:

Errors handling
---------------

Both consumer and producer can raise exceptions that inherit from the `kafka.common.KafkaError` class
and declared in `kafka.common` module.

Example of exceptions handling:


.. code:: python
from kafka.common import KafkaError, KafkaTimeoutError
# ...
try:
send_future = yield from producer.send('foobar', b'test data')
response = yield from send_future # wait until message is produced
except KafkaTimeourError:
print("produce timeout... maybe we want to resend data again?")
except KafkaError as err:
print("some kafka error on produce: {}".format(err))

0 comments on commit 2d95376

Please sign in to comment.