Repository: infinyon/fluvio-client-python (language: python)

Issue:
Python Iterator Cleanup
Right now, to use a `PartitionConsumerStream` as a [python iterator](https://wiki.python.org/moin/Iterator), we do a work around of creating a python class [`PartitionConsumerStreamIterator`](https://github.com/infinyon/fluvio-client-python/blob/a1c6b369177324105d1c2372a0e6c558e4b24f10/fluvio/__init__.py#L3-L11):
```python
class PartitionConsumerStreamIterator:
    def __init__(self, stream):
        self.stream = stream

    def __iter__(self):
        return self

    def __next__(self):
        return self.stream.next()
```
where stream is the `PartitionConsumerStream`. It'd be more ideal if `PartitionConsumerStream` had the `__iter__` and `__next__` methods.
Add Key Value Records
Requires adding a `Record` type to the python side.

Related:
* infinyon/fluvio#828.
* infinyon/fluvio-client-node#77


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 86160286..ba797243 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -22,7 +22,7 @@ jobs:
       matrix:
         os: [ubuntu-latest, macos-latest]
         rust: [stable]
-        python-version: [3.8]
+        python-version: [3.9]
     steps:
       - uses: actions/checkout@v2
       - name: Set up Python ${{ matrix.python-version }}
@@ -36,8 +36,8 @@ jobs:
           toolchain: ${{ matrix.rust }}
           override: true
           components: clippy
-      - name: check fmt
-        run: cargo fmt -- --check
+      - name: check fmt and flake8
+        run: make lint
   smoke_test:
     name: Smoke test
     runs-on: ${{ matrix.os }}
@@ -64,6 +64,7 @@ jobs:
         uses: infinyon/fluvio@master
         with:
           cluster-type: local
+          version: latest
       - name: Check Fluvio Installation
         run: |
           fluvio version
diff --git a/Cargo.lock b/Cargo.lock
index 2a12c87e..5760a735 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1,5 +1,17 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
+[[package]]
+name = "_fluvio_python"
+version = "0.0.2"
+dependencies = [
+ "cpython",
+ "env_logger",
+ "flapigen",
+ "fluvio",
+ "fluvio-future",
+ "log",
+]
+
 [[package]]
 name = "aho-corasick"
 version = "0.7.15"
@@ -186,9 +198,9 @@ checksum = "e91831deabf0d6d7ec49552e489aed63b7456a7a3c46cff62adad428110b0af0"
 
 [[package]]
 name = "async-trait"
-version = "0.1.42"
+version = "0.1.48"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8d3a45e77e34375a7923b1e8febb049bb011f064714a8e17a1a616fef01da13d"
+checksum = "36ea56748e10732c49404c153638a15ec3d6211ec5ff35d9bb20e13b93576adf"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -378,9 +390,9 @@ dependencies = [
 
 [[package]]
 name = "ctor"
-version = "0.1.15"
+version = "0.1.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "39858aa5bac06462d4dd4b9164848eb81ffc4aa5c479746393598fd193afa227"
+checksum = "e8f45d9ad417bcef4817d614a501ab55cdd96a6fdb24f49aab89a54acfd66b19"
 dependencies = [
  "quote",
  "syn",
@@ -445,7 +457,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "rustc-hash",
- "smallvec 1.4.2",
+ "smallvec 1.6.1",
  "smol_str",
  "strum",
  "syn",
@@ -454,9 +466,8 @@ dependencies = [
 
 [[package]]
 name = "fluvio"
-version = "0.5.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "284cfcbb6b412e646a4caa4a4db9daf6a7c67a7ff3a9aae2ec1521c566e7ca23"
+version = "0.6.0"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "async-channel",
  "async-mutex",
@@ -488,8 +499,7 @@ dependencies = [
 [[package]]
 name = "fluvio-controlplane-metadata"
 version = "0.6.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "756c5bd4f47db68ebd316a9866e1fb9f8e1d584b5b9effa278f2026b3654074e"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "async-trait",
  "fluvio-dataplane-protocol",
@@ -505,8 +515,7 @@ dependencies = [
 [[package]]
 name = "fluvio-dataplane-protocol"
 version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "67d95c8bdc1d26c1625ab3dea79bc9921a9bb738edac75781afe8fe56422ab8f"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "bytes",
  "content_inspector",
@@ -544,8 +553,7 @@ dependencies = [
 [[package]]
 name = "fluvio-protocol"
 version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d06bb9a34769434a029c4c6424b5cf104e6372e619e9da1c49bf5c6ffbfe6b3"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "bytes",
  "fluvio-future",
@@ -559,8 +567,7 @@ dependencies = [
 [[package]]
 name = "fluvio-protocol-api"
 version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4997e09168ef29658b02986e570ac1c483536df2775da134c60c617ba53a7425"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "fluvio-protocol-core",
  "fluvio-protocol-derive",
@@ -571,8 +578,7 @@ dependencies = [
 [[package]]
 name = "fluvio-protocol-codec"
 version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3aa87d4b153d262a270aa8dea9ef2f816f512ad4decf4c4048e1890ce9c9f26f"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "bytes",
  "fluvio-protocol-core",
@@ -583,8 +589,7 @@ dependencies = [
 [[package]]
 name = "fluvio-protocol-core"
 version = "0.2.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "60265c4636c4de93156d25aa9ae0869e2133009733f50c3cef57886ca0641ebe"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "bytes",
  "log",
@@ -593,8 +598,7 @@ dependencies = [
 [[package]]
 name = "fluvio-protocol-derive"
 version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c1a2bb5eeabcfffd406155caa7dfc1e75367c6dd825c769572815f6300f9c50f"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "log",
  "proc-macro2",
@@ -602,23 +606,10 @@ dependencies = [
  "syn",
 ]
 
-[[package]]
-name = "fluvio-python"
-version = "0.0.1"
-dependencies = [
- "cpython",
- "env_logger",
- "flapigen",
- "fluvio",
- "fluvio-future",
- "log",
-]
-
 [[package]]
 name = "fluvio-sc-schema"
 version = "0.6.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "48d222791d9f7c3849e275d71894ed957e3958b56d9f02d7374b83cb70d01016"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "fluvio-controlplane-metadata",
  "fluvio-dataplane-protocol",
@@ -633,8 +624,7 @@ dependencies = [
 [[package]]
 name = "fluvio-socket"
 version = "0.6.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ebc2793895b46f1b5938a368a22108b9cdc6ee1840da4cfdb1b8a1e5b9eb503"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "async-channel",
  "async-mutex",
@@ -659,8 +649,7 @@ dependencies = [
 [[package]]
 name = "fluvio-spu-schema"
 version = "0.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0f6e134642621f124238207cc9f35924ebab0479d2aff35e6b18ae0c6ec71b3d"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "bytes",
  "fluvio-dataplane-protocol",
@@ -674,8 +663,7 @@ dependencies = [
 [[package]]
 name = "fluvio-stream-model"
 version = "0.4.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c8a144509519dcd5d9ca82a36b44419e14497176402125065dffe31eda9a091"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "async-rwlock",
  "event-listener",
@@ -685,9 +673,8 @@ dependencies = [
 
 [[package]]
 name = "fluvio-types"
-version = "0.2.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "591ae978fa5451502e6065fbc717ef92a334d9281ec8179caf21fe2c7c8adf34"
+version = "0.2.3"
+source = "git+https://github.com/infinyon/fluvio?rev=b7b3d2c16880ea21f1c67246ce9765368c991ebe#b7b3d2c16880ea21f1c67246ce9765368c991ebe"
 dependencies = [
  "event-listener",
  "tracing",
@@ -894,9 +881,9 @@ dependencies = [
 
 [[package]]
 name = "indexmap"
-version = "1.6.1"
+version = "1.6.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4fb1fa934250de4de8aef298d81c729a7d33d8c239daa3a7575e6b92bfc7313b"
+checksum = "824845a0bf897a9042383849b02c1bc219c2383772efcd5c6f9766fa4b81aef3"
 dependencies = [
  "autocfg",
  "hashbrown",
@@ -904,9 +891,12 @@ dependencies = [
 
 [[package]]
 name = "instant"
-version = "0.1.6"
+version = "0.1.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5b141fdc7836c525d4d594027d318c84161ca17aaf8113ab1f81ab93ae897485"
+checksum = "61124eeebbd69b8190558df225adf7e4caafce0d743919e5d6b19652314ec5ec"
+dependencies = [
+ "cfg-if 1.0.0",
+]
 
 [[package]]
 name = "itoa"
@@ -940,9 +930,9 @@ checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646"
 
 [[package]]
 name = "libc"
-version = "0.2.77"
+version = "0.2.88"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f2f96b10ec2560088a8e76961b00d47107b3a625fecb76dedb29ee7ccbf98235"
+checksum = "03b07a082330a35e43f63177cc01689da34fbffa0105e1246cf0311472cac73a"
 
 [[package]]
 name = "log"
@@ -1254,9 +1244,9 @@ checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0"
 
 [[package]]
 name = "quote"
-version = "1.0.7"
+version = "1.0.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aa563d17ecb180e500da1cfd2b028310ac758de548efdd203e18f283af693f37"
+checksum = "c3d0b9745dc2debf507c8422de05d7226cc1f0644216dfdfead988f9b1ab32a7"
 dependencies = [
  "proc-macro2",
 ]
@@ -1366,21 +1356,20 @@ dependencies = [
 
 [[package]]
 name = "regex"
-version = "1.4.3"
+version = "1.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9251239e129e16308e70d853559389de218ac275b515068abc96829d05b948a"
+checksum = "54fd1046a3107eb58f42de31d656fee6853e5d276c455fd943742dce89fc3dd3"
 dependencies = [
  "aho-corasick",
  "memchr",
  "regex-syntax",
- "thread_local",
 ]
 
 [[package]]
 name = "regex-syntax"
-version = "0.6.22"
+version = "0.6.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b5eb417147ba9860a96cfe72a0b93bf88fee1744b5636ec99ab20c1aa9376581"
+checksum = "24d5f089152e60f62d28b835fbff2cd2e8dc0baf1ac13343bef92ab7eed84548"
 
 [[package]]
 name = "remove_dir_all"
@@ -1427,9 +1416,9 @@ dependencies = [
 
 [[package]]
 name = "security-framework"
-version = "2.1.1"
+version = "2.1.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2dfd318104249865096c8da1dfabf09ddbb6d0330ea176812a62ec75e40c4166"
+checksum = "d493c5f39e02dfb062cd8f33301f90f9b13b650e8c1b1d0fd75c19dd64bff69d"
 dependencies = [
  "bitflags",
  "core-foundation",
@@ -1468,18 +1457,18 @@ dependencies = [
 
 [[package]]
 name = "serde"
-version = "1.0.123"
+version = "1.0.124"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "92d5161132722baa40d802cc70b15262b98258453e85e5d1d365c757c73869ae"
+checksum = "bd761ff957cb2a45fbb9ab3da6512de9de55872866160b23c25f1a841e99d29f"
 dependencies = [
  "serde_derive",
 ]
 
 [[package]]
 name = "serde_derive"
-version = "1.0.123"
+version = "1.0.124"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9391c295d64fc0abb2c556bad848f33cb8296276b1ad2677d1ae1ace4f258f31"
+checksum = "1800f7693e94e186f5e25a28291ae1570da908aff7d97a095dec1e56ff99069b"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1499,9 +1488,9 @@ dependencies = [
 
 [[package]]
 name = "signal-hook"
-version = "0.3.6"
+version = "0.3.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8a7f3f92a1da3d6b1d32245d0cbcbbab0cfc45996d8df619c42bccfa6d2bbb5f"
+checksum = "6aa894ef3fade0ee7243422f4fbbd6c2b48e6de767e621d37ef65f2310f53cea"
 dependencies = [
  "libc",
  "signal-hook-registry",
@@ -1533,9 +1522,9 @@ dependencies = [
 
 [[package]]
 name = "smallvec"
-version = "1.4.2"
+version = "1.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fbee7696b84bbf3d89a1c2eccff0850e3047ed46bfcd2e92c29a2d074d57e252"
+checksum = "fe0f37c9e8f3c5a4a66ad655a93c74daac4ad00c441533bf5c6e7990bb42604e"
 
 [[package]]
 name = "smol_str"
@@ -1589,9 +1578,9 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "1.0.60"
+version = "1.0.63"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c700597eca8a5a762beb35753ef6b94df201c81cca676604f547495a0d7f0081"
+checksum = "8fd9bc7ccc2688b3344c2f48b9b546648b25ce0b20fc717ee7fa7981a8ca9717"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -1641,15 +1630,6 @@ dependencies = [
  "syn",
 ]
 
-[[package]]
-name = "thread_local"
-version = "1.1.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8018d24e04c95ac8790716a5987d0fec4f8b27249ffa0f7d33f1369bdfb88cbd"
-dependencies = [
- "once_cell",
-]
-
 [[package]]
 name = "tinyvec"
 version = "1.1.1"
@@ -1726,9 +1706,9 @@ dependencies = [
 
 [[package]]
 name = "tracing-attributes"
-version = "0.1.13"
+version = "0.1.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a8a9bd1db7706f2373a190b0d067146caa39350c486f3d455b0e33b431f94c07"
+checksum = "41768be5b9f3489491825f56f01f25290aa1d3e7cc97e182d4d34360493ba6fa"
 dependencies = [
  "proc-macro2",
  "quote",
diff --git a/Cargo.toml b/Cargo.toml
index 891d7aec..2fc9a987 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,12 +1,12 @@
 [package]
-name = "fluvio-python"
-version = "0.0.1"
+name = "_fluvio_python"
+version = "0.0.2"
 edition = "2018"
 authors = ["Fluvio Contributors <team@fluvio.io>"]
 
 [lib]
 crate-type = ["cdylib"]
-name = "fluvio_python"
+name = "_fluvio_python"
 
 [build-dependencies]
 flapigen = "0.6.0-pre7"
@@ -15,5 +15,6 @@ log = "^0.4.6"
 
 [dependencies]
 cpython = { version = "0.5", features = ["extension-module"] }
-fluvio = { version = "0.5" }
+#fluvio = { version = "0.5" }
+fluvio = { git = "https://github.com/infinyon/fluvio", rev = "b7b3d2c16880ea21f1c67246ce9765368c991ebe" }
 fluvio-future = { vesrion = "0.1.15", features = ["task", "io"] }
diff --git a/Makefile b/Makefile
index 33324b0b..c809ddae 100644
--- a/Makefile
+++ b/Makefile
@@ -7,32 +7,37 @@ venv:
 	python3 -m venv venv
 
 venv-pip: venv
-	$(PIP) install -U pip setuptools
+	$(PIP) install -U pip setuptools pdoc flake8 ipdb
 	$(PIP) install -r requirements.txt
 
+lint: venv-pip
+	cargo fmt -- --check
+	$(PYTHON) -m flake8 fluvio tests
+
 build-wheel: venv-pip
+	rm -rf ./fluvio.egg-info/
 	$(PYTHON) setup.py bdist_wheel
 
 install-wheel: build-wheel
-	#rm -r ./venv/lib/python3.9/site-packages/fluvio*
 	$(PIP) install --upgrade --force-reinstall --no-index --pre --find-links=dist/ fluvio
 
 build-dev: venv-pip
-	#$(PYTHON) setup.py develop
-	$(PYTHON) setup.py install
+	$(PYTHON) setup.py develop
 
 test: install-wheel
 	fluvio topic create my-topic-iterator || true
-	fluvio topic create my-topic-while || true
 	fluvio topic create my-topic-produce || true
+	fluvio topic create my-topic-key-value-iterator || true
 	cd tests && ../venv/bin/python -m unittest
 	fluvio topic delete my-topic-iterator || true
-	fluvio topic delete my-topic-while || true
 	fluvio topic delete my-topic-produce || true
+	fluvio topic delete my-topic-key-value-iterator || true
 
 ci-build: venv-pip
 	CIBW_SKIP="cp27-*" $(PYTHON) -m cibuildwheel --platform linux --output-dir wheelhouse
 
+docs-serve: venv-pip build-dev
+	$(PYTHON) -m pdoc fluvio
 
 clean:
-	rm -rf venv fluvio/*.so target
+	rm -rf venv fluvio/*.so target dist build
diff --git a/README.md b/README.md
index bd67f47a..6fbc16ac 100644
--- a/README.md
+++ b/README.md
@@ -6,7 +6,10 @@
 </div>
 <br />
 
-[![Build Status](https://github.com/infinyon/fluvio-client-python/workflows/CI/badge.svg)](https://github.com/infinyon/flv-client-python/actions) [![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://github.com/infinyon/flv-client-python/blob/master/LICENSE-APACHE) [![PyPi](https://img.shields.io/pypi/v/fluvio.svg)](https://img.shields.io/pypi/v/fluvio.svg)
+[![Build
+Status](https://github.com/infinyon/fluvio-client-python/workflows/CI/badge.svg)](https://github.com/infinyon/flv-client-python/actions)
+[![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://github.com/infinyon/flv-client-python/blob/master/LICENSE-APACHE)
+[![PyPi](https://img.shields.io/pypi/v/fluvio.svg)](https://img.shields.io/pypi/v/fluvio.svg)
 
 # Usage
 
@@ -24,21 +27,12 @@ from fluvio import Fluvio
 fluvio = Fluvio.connect()
 consumer = fluvio.partition_consumer('my-topic-while', 0)
 stream = consumer.stream(0)
-```
-Given the `stream` instance, you can use it by calling `.next()` like this:
-```python
-curr = stream.next()
-while curr is not None:
-    print(curr)
-```
 
-Or we've added an iterator wrapped around it:
-```python
-for i in PartitionConsumerStreamIterator(stream):
-    print(i)
+for i in stream:
+    print(i.value_string())
 ```
 
-# Development Notes
+# Developer Notes
 
 This project uses [flapigen](https://github.com/Dushistov/flapigen-rs) to
 genate the C static library and
diff --git a/build.rs b/build.rs
index 0c13cbdd..449aacfa 100644
--- a/build.rs
+++ b/build.rs
@@ -7,7 +7,7 @@ fn main() {
     let out_dir = env::var("OUT_DIR").unwrap();
     let out_src = Path::new(&out_dir).join("glue.rs");
 
-    let python_cfg = PythonConfig::new("fluvio_python".to_owned());
+    let python_cfg = PythonConfig::new("_fluvio_python".to_owned());
     let flap_gen =
         flapigen::Generator::new(LanguageConfig::PythonConfig(python_cfg)).rustfmt_bindings(true);
     flap_gen.expand("python bindings", &in_src, &out_src);
diff --git a/fluvio/__init__.py b/fluvio/__init__.py
index 9e89e991..0afb15bd 100644
--- a/fluvio/__init__.py
+++ b/fluvio/__init__.py
@@ -1,11 +1,177 @@
-from .fluvio_python import *
+from ._fluvio_python import (
+    Fluvio as _Fluvio,
+    PartitionConsumer as _PartitionConsumer,
+    PartitionConsumerStream as _PartitionConsumerStream,
+    TopicProducer as _TopicProducer,
+    Record as _Record,
+)
+from ._fluvio_python import Error as FluviorError  # noqa: F401
+import typing
 
-class PartitionConsumerStreamIterator:
-    def __init__(self, stream):
-        self.stream = stream
+
+class Record:
+    '''The individual record for a given stream.
+    '''
+    _inner: _Record
+
+    def __init__(self, inner: _Record):
+        self._inner = inner
+
+    def offset(self) -> int:
+        '''The offset from the initial offset for a given stream.
+        '''
+        return self._inner.offset()
+
+    def value(self) -> typing.List[int]:
+        '''Returns the contents of this Record's value
+        '''
+        return self._inner.value()
+
+    def value_string(self) -> str:
+        '''The UTF-8 decoded value for this record.
+        '''
+        return self._inner.value_string()
+
+    def key(self) -> typing.List[int]:
+        '''Returns the contents of this Record's key, if it exists
+        '''
+        return self._inner.key()
+
+    def key_string(self) -> str:
+        '''The UTF-8 decoded key for this record.
+        '''
+        return self._inner.key_string()
+
+
+class PartitionConsumerStream:
+    '''An iterator for `PartitionConsumer.stream` method where each `__next__`
+    returns a `Record`.
+
+    Usage:
+
+    ```python
+    for i in consumer.stream(0):
+        print(i.value())
+        print(i.value_string())
+    ```
+    '''
+    _inner: _PartitionConsumerStream
+
+    def __init__(self, inner: _PartitionConsumerStream):
+        self._inner = inner
 
     def __iter__(self):
         return self
 
-    def __next__(self):
-        return self.stream.next()
+    def __next__(self) -> typing.Optional[Record]:
+        return Record(self._inner.next())
+
+
+class PartitionConsumer:
+    '''
+    An interface for consuming events from a particular partition
+
+    There are two ways to consume events: by "fetching" events and by
+    "streaming" events. Fetching involves specifying a range of events that you
+    want to consume via their Offset. A fetch is a sort of one-time batch
+    operation: you’ll receive all of the events in your range all at once. When
+    you consume events via Streaming, you specify a starting Offset and receive
+    an object that will continuously yield new events as they arrive.
+    '''
+
+    _inner: _PartitionConsumer
+
+    def __init__(self, inner: _PartitionConsumer):
+        self._inner = inner
+
+    def stream(self, offset: int) -> PartitionConsumerStream:
+        '''
+        Continuously streams events from a particular offset in the consumer’s
+        partition. This returns a `PartitionConsumerStream` which is an
+        iterator.
+
+        Streaming is one of the two ways to consume events in Fluvio. It is a
+        continuous request for new records arriving in a partition, beginning
+        at a particular offset. You specify the starting point of the stream
+        using an Offset and periodically receive events, either individually or
+        in batches.
+        '''
+        return PartitionConsumerStream(self._inner.stream(offset))
+
+
+class TopicProducer:
+    '''An interface for producing events to a particular topic.
+
+    A `TopicProducer` allows you to send events to the specific topic it was
+    initialized for. Once you have a `TopicProducer`, you can send events to
+    the topic, choosing which partition each event should be delivered to.
+    '''
+    _inner: _TopicProducer
+
+    def __init__(self, inner: _TopicProducer):
+        self._inner = inner
+
+    def send_record_string(self, buf: str, partition: int) -> None:
+        '''Sends an event to a specific partition within this producer’s topic
+        '''
+        return self._inner.send_record(buf.encode('utf-8'), partition)
+
+    def send_record(self, buf: typing.List[int], partition: int) -> None:
+        '''Sends an event to a specific partition within this producer’s topic
+        '''
+        return self._inner.send_record(buf, partition)
+
+    def send(self, key: typing.List[int], value: typing.List[int]) -> None:
+        '''
+        Sends a key/value record to this producer's Topic.
+
+        The partition that the record will be sent to is derived from the Key.
+        '''
+        return self._inner.send(key, value)
+
+
+class Fluvio:
+    '''An interface for interacting with Fluvio streaming.'''
+    _inner: _Fluvio
+
+    def __init__(self, inner: _Fluvio):
+        self._inner = inner
+
+    @classmethod
+    def connect(cls):
+        '''Creates a new Fluvio client using the current profile from
+        `~/.fluvio/config`
+
+        If there is no current profile or the `~/.fluvio/config` file does not
+        exist, then this will create a new profile with default settings and
+        set it as current, then try to connect to the cluster using those
+        settings.
+        '''
+        return cls(_Fluvio.connect())
+
+    def partition_consumer(
+        self,
+        topic: str,
+        partition: int
+    ) -> PartitionConsumer:
+        '''Creates a new `PartitionConsumer` for the given topic and partition
+
+        Currently, consumers are scoped to both a specific Fluvio topic and to
+        a particular partition within that topic. That means that if you have a
+        topic with multiple partitions, then in order to receive all of the
+        events in all of the partitions, you will need to create one consumer
+        per partition.
+        '''
+        return PartitionConsumer(
+            self._inner.partition_consumer(topic, partition)
+        )
+
+    def topic_producer(self, topic: str) -> TopicProducer:
+        '''
+        Creates a new `TopicProducer` for the given topic name.
+
+        Currently, producers are scoped to a specific Fluvio topic. That means
+        when you send events via a producer, you must specify which partition
+        each event should go to.
+        '''
+        return TopicProducer(self._inner.topic_producer(topic))
diff --git a/setup.py b/setup.py
index 9a0b1390..072f0a95 100644
--- a/setup.py
+++ b/setup.py
@@ -16,7 +16,6 @@
     setup_requires=['wheel'],
     project_urls={  # Optional
         'Bug Reports': 'https://github.com/infinyon/fluvio-client-python/issues',
-        'Say Thanks!': 'http://saythanks.io/to/example',
         'Source': 'https://github.com/infinyon/fluvio-client-python',
     },
     # For a list of valid classifiers, see https://pypi.org/classifiers/
@@ -38,7 +37,7 @@
         'Programming Language :: Python :: 3 :: Only',
     ],
 
-    rust_extensions=[RustExtension("fluvio.fluvio_python", path="Cargo.toml", binding=Binding.RustCPython, debug=False)],
+    rust_extensions=[RustExtension("fluvio._fluvio_python", path="Cargo.toml", binding=Binding.RustCPython, debug=False)],
     packages=["fluvio"],
     # rust extensions are not zip safe, just like C-extensions.
     zip_safe=False,
diff --git a/src/glue.rs.in b/src/glue.rs.in
index 2d6535c9..5bf6b422 100644
--- a/src/glue.rs.in
+++ b/src/glue.rs.in
@@ -20,7 +20,11 @@ mod _Fluvio {
     pub fn connect() -> Fluvio {
         run_block_on(Fluvio::connect()).expect("Failed to connect to fluvio")
     }
-    pub fn partition_consumer(fluvio: &Fluvio, topic: String, partition: i32) -> Result<PartitionConsumer, FluvioError> {
+    pub fn partition_consumer(
+        fluvio: &Fluvio,
+        topic: String,
+        partition: i32
+    ) -> Result<PartitionConsumer, FluvioError> {
         run_block_on(fluvio.partition_consumer(topic, partition))
     }
     pub fn topic_producer(
@@ -35,8 +39,15 @@ foreign_class!(class Fluvio {
     self_type Fluvio;
     private constructor = empty;
     fn _Fluvio::connect() -> Fluvio;
-    fn _Fluvio::partition_consumer(&self, _: String, _ : i32) -> Result<PartitionConsumer, FluvioError>;
-    fn _Fluvio::topic_producer(&self, _: String) -> Result<TopicProducer, FluvioError>;
+    fn _Fluvio::partition_consumer(
+        &self,
+        _: String,
+        _ : i32
+    ) -> Result<PartitionConsumer, FluvioError>;
+    fn _Fluvio::topic_producer(
+        &self,
+        _: String
+    ) -> Result<TopicProducer, FluvioError>;
 });
 
 
@@ -47,14 +58,18 @@ mod _PartitionConsumer {
         offset: u32,
     ) -> Result<PartitionConsumerStream, FluvioError> {
         Ok(PartitionConsumerStream {
-            inner: Box::pin(run_block_on(consumer.stream(Offset::from_beginning(offset)))?)
+            inner: Box::pin(run_block_on(
+               consumer.stream(Offset::beginning())
+            )?)
         })
     }
 }
 foreign_class!(class PartitionConsumer {
     self_type PartitionConsumer;
     private constructor = empty;
-    fn _PartitionConsumer::stream(&self, _: u32) -> Result<PartitionConsumerStream, FluvioError>;
+    fn _PartitionConsumer::stream(
+        &self, _: u32
+    ) -> Result<PartitionConsumerStream, FluvioError>;
 });
 
 type PartitionConsumerIteratorInner =
@@ -64,39 +79,72 @@ pub struct PartitionConsumerStream {
     pub inner: PartitionConsumerIteratorInner,
 }
 impl PartitionConsumerStream {
-    pub fn next(&mut self) -> Result<Option<String>, FluvioError> {
-         use std::convert::TryFrom;
-        if let Some(record) = run_block_on(self.inner.next()) {
-            let record = record?;
-            if let Ok(record) = String::try_from(record) {
-                Ok(Some(record))
-            } else {
-                Ok(None)
-            }
-        } else {
-            Ok(None)
-        }
+    pub fn next(&mut self) -> Option<Result<Record, FluvioError>> {
+        run_block_on(self.inner.next())
     }
 }
 
 foreign_class!(class PartitionConsumerStream{
     self_type PartitionConsumerStream;
     private constructor = empty;
-    fn PartitionConsumerStream::next(&mut self) -> Result<Option<String>, FluvioError>;
+    fn PartitionConsumerStream::next(
+        &mut self
+    ) -> Option<Result<Record, FluvioError>>;
 });
 
 mod _TopicProducer {
     use super::*;
     pub fn send_record(
         producer: &TopicProducer,
-        buffer: String,
+        buffer: &[u8],
         partition: i32,
     ) -> Result<(), FluvioError> {
         run_block_on(producer.send_record(buffer, partition))
     }
+    pub fn send(
+        producer: &TopicProducer,
+        key: &[u8],
+        value: &[u8],
+    ) -> Result<(), FluvioError> {
+        run_block_on(producer.send(key, value))
+    }
 }
 foreign_class!(class TopicProducer {
     self_type TopicProducer;
     private constructor = empty;
-    fn _TopicProducer::send_record(&self, _: String, _ : i32) -> Result<(), FluvioError>;
+    fn _TopicProducer::send_record(
+        &self,
+        _: &[u8],
+        _ : i32
+    ) -> Result<(), FluvioError>;
+    fn _TopicProducer::send(
+        &self,
+        _: &[u8],
+        _ : &[u8]
+    ) -> Result<(), FluvioError>;
+});
+
+
+use std::string::FromUtf8Error;
+mod _Record {
+    use super::*;
+    pub fn value_string(record: &Record) -> Result<String, FromUtf8Error> {
+        String::from_utf8(record.value().to_vec())
+    }
+    pub fn key_string(
+        record: &Record
+    ) -> Option<Result<String, FromUtf8Error>> {
+        let key = record.key()?;
+        Some(String::from_utf8(key.to_vec()))
+    }
+}
+
+foreign_class!(class Record {
+    self_type Record;
+    private constructor = empty;
+    fn Record::offset(&self) -> i64;
+    fn Record::value(&self) -> &[u8];
+    fn Record::key(&self) -> Option<&[u8]>;
+    fn _Record::value_string(&self) -> Result<String, FromUtf8Error>;
+    fn _Record::key_string(&self) -> Option<Result<String, FromUtf8Error>>;
 });


Test patch:
diff --git a/tests/test_fluvio_python.py b/tests/test_fluvio_python.py
index 84521eca..f7f50338 100644
--- a/tests/test_fluvio_python.py
+++ b/tests/test_fluvio_python.py
@@ -1,42 +1,73 @@
-from fluvio import Fluvio, PartitionConsumerStreamIterator
+from fluvio import (Fluvio, FluviorError)
 import unittest
 
+
 class TestFluvioMethods(unittest.TestCase):
     def test_connect(self):
-        fluvio = Fluvio.connect()
+        # A very simple test
+        Fluvio.connect()
 
     def test_produce(self):
         fluvio = Fluvio.connect()
 
         producer = fluvio.topic_producer('my-topic-produce')
         for i in range(10):
-            producer.send_record("FOOBAR %s " % i, 0)
+            producer.send_record_string("FOOBAR %s " % i, 0)
 
-    def test_consume_with_while_loop(self):
+    def test_produce_on_uncreated_topic(self):
         fluvio = Fluvio.connect()
-        producer = fluvio.topic_producer('my-topic-while')
-        for i in range(10):
-            producer.send_record("FOOBAR %s " % i, 0)
-
-        consumer = fluvio.partition_consumer('my-topic-while', 0)
-        stream = consumer.stream(0)
-        count = 1
-        curr = stream.next()
-        while curr is not None and count < 10:
-            print(curr)
-            curr = stream.next()
-            count += 1
+
+        producer = fluvio.topic_producer('a-topic-that-does-not-exist')
+        try:
+            producer.send_record_string("THIS SHOULD FAIL", 0)
+        except FluviorError as e:
+
+            self.assertEqual(
+                e.args,
+                ('Partition not found: a-topic-that-does-not-exist-0',)
+            )
+            print('ERROR: %s' % e)
 
     def test_consume_with_iterator(self):
         fluvio = Fluvio.connect()
         producer = fluvio.topic_producer('my-topic-iterator')
         for i in range(10):
-            producer.send_record("FOOBAR %s " % i, 0)
+            producer.send_record_string("record-%s" % i, 0)
 
         consumer = fluvio.partition_consumer('my-topic-iterator', 0)
         count = 0
-        for i in PartitionConsumerStreamIterator(consumer.stream(0)):
-            print("THIS IS IN AN ITERATOR! %s" % i)
+        for i in consumer.stream(0):
+            print("THIS IS IN AN ITERATOR! %s" % i.value())
+            self.assertEqual(
+                bytearray(i.value()).decode(), 'record-%s' % count
+            )
+            self.assertEqual(i.value_string(), 'record-%s' % count)
+            count += 1
+            if count >= 10:
+                break
+
+    def test_key_value(self):
+        fluvio = Fluvio.connect()
+        producer = fluvio.topic_producer('my-topic-key-value-iterator')
+        for i in range(10):
+            producer.send("foo".encode(), ("record-%s" % i).encode())
+
+        consumer = fluvio.partition_consumer('my-topic-key-value-iterator', 0)
+        count = 0
+        for i in consumer.stream(0):
+            print(
+                "THIS IS IN AN ITERATOR! key - %s, value - %s" % (
+                    i.key(),
+                    i.value()
+                )
+            )
+            self.assertEqual(
+                bytearray(i.value()).decode(), 'record-%s' % count
+            )
+            self.assertEqual(i.value_string(), 'record-%s' % count)
+            self.assertEqual(i.key_string(), 'foo')
+            self.assertEqual(i.key(), list('foo'.encode()))
+
             count += 1
             if count >= 10:
                 break