diff --git a/dependency-versions.gradle b/dependency-versions.gradle
index 9c3d571..fa8d1cd 100644
--- a/dependency-versions.gradle
+++ b/dependency-versions.gradle
@@ -29,6 +29,7 @@
     dependency('info.picocli:picocli:4.0.0-alpha-2')
     dependency('io.lettuce:lettuce-core:5.1.3.RELEASE')
     dependency('io.vertx:vertx-core:3.6.2')
+    dependency('javax.persistence:javax.persistence-api:2.2')
     dependencySet(group: 'org.antlr', version: '4.7.1') {
       entry 'antlr4'
       entry 'antlr4-runtime'
diff --git a/eth-repository/src/main/kotlin/org/apache/tuweni/eth/repository/BlockchainRepository.kt b/eth-repository/src/main/kotlin/org/apache/tuweni/eth/repository/BlockchainRepository.kt
index b7c478e..9c4e30a 100644
--- a/eth-repository/src/main/kotlin/org/apache/tuweni/eth/repository/BlockchainRepository.kt
+++ b/eth-repository/src/main/kotlin/org/apache/tuweni/eth/repository/BlockchainRepository.kt
@@ -41,10 +41,10 @@
  * @param blockchainIndex the blockchain index to index values
  */
   (
-    private val chainMetadata: KeyValueStore,
-    private val blockBodyStore: KeyValueStore,
-    private val blockHeaderStore: KeyValueStore,
-    private val transactionReceiptsStore: KeyValueStore,
+    private val chainMetadata: KeyValueStore<Bytes, Bytes>,
+    private val blockBodyStore: KeyValueStore<Bytes, Bytes>,
+    private val blockHeaderStore: KeyValueStore<Bytes, Bytes>,
+    private val transactionReceiptsStore: KeyValueStore<Bytes, Bytes>,
     private val blockchainIndex: BlockchainIndex
   ) {
 
@@ -58,10 +58,10 @@
      * @return a new blockchain repository made from the metadata passed in parameter.
      */
     suspend fun init(
-      blockBodyStore: KeyValueStore,
-      blockHeaderStore: KeyValueStore,
-      chainMetadata: KeyValueStore,
-      transactionReceiptsStore: KeyValueStore,
+      blockBodyStore: KeyValueStore<Bytes, Bytes>,
+      blockHeaderStore: KeyValueStore<Bytes, Bytes>,
+      chainMetadata: KeyValueStore<Bytes, Bytes>,
+      transactionReceiptsStore: KeyValueStore<Bytes, Bytes>,
       blockchainIndex: BlockchainIndex,
       genesisBlock: Block
     ): BlockchainRepository {
diff --git a/kv/build.gradle b/kv/build.gradle
index 0f20ad0..d05b5c2 100644
--- a/kv/build.gradle
+++ b/kv/build.gradle
@@ -19,6 +19,7 @@
   compile 'org.jetbrains.kotlinx:kotlinx-coroutines-guava'
   compile 'org.jetbrains.kotlinx:kotlinx-coroutines-jdk8'
   compile 'org.jetbrains.kotlin:kotlin-stdlib-jdk8'
+  compile 'javax.persistence:javax.persistence-api'
   compileOnly 'com.jolbox:bonecp'
   compileOnly 'io.lettuce:lettuce-core'
   compileOnly 'org.fusesource.leveldbjni:leveldbjni-all'
diff --git a/kv/src/main/kotlin/org/apache/tuweni/kv/EntityManagerKeyValueStore.kt b/kv/src/main/kotlin/org/apache/tuweni/kv/EntityManagerKeyValueStore.kt
new file mode 100644
index 0000000..8b653ad
--- /dev/null
+++ b/kv/src/main/kotlin/org/apache/tuweni/kv/EntityManagerKeyValueStore.kt
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * 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.
+ */
+package org.apache.tuweni.kv
+
+import kotlin.jvm.Throws
+import kotlinx.coroutines.Dispatchers
+
+import java.io.IOException
+import java.util.function.Function
+import java.util.function.Supplier
+import java.util.stream.Stream
+import javax.persistence.EntityManager
+import kotlin.coroutines.CoroutineContext
+
+class EntityManagerKeyValueStore<K, V>
+        @Throws(IOException::class)
+        constructor(
+          private val entityManagerProvider: () -> EntityManager,
+          private val entityClass: Class<V>,
+          private val idAccessor: (V) -> K,
+          override val coroutineContext: CoroutineContext = Dispatchers.IO
+        ) : KeyValueStore<K, V> {
+
+  companion object {
+    /**
+     * Open a relational database backed key-value store using a JPA entity manager.
+     *
+     * @param entityManagerProvider The supplier of entity manager to operate.
+     * @return A key-value store.
+     * @throws IOException If an I/O error occurs.
+     */
+    @JvmStatic
+    @Throws(IOException::class)
+    fun <K, V> open(
+      entityManagerProvider: Supplier<EntityManager>,
+      entityClass: Class<V>,
+      idAccessor: Function<V, K>
+    ) = EntityManagerKeyValueStore<K, V>(entityManagerProvider::get, entityClass, idAccessor::apply)
+  }
+
+  override suspend fun get(key: K): V? {
+    val em = entityManagerProvider()
+    em.transaction.begin()
+    try {
+      return em.find(entityClass, key)
+    } finally {
+      em.transaction.commit()
+      em.close()
+    }
+  }
+
+  override suspend fun put(key: K, value: V) {
+    val em = entityManagerProvider()
+    em.transaction.begin()
+    try {
+      em.merge(value)
+    } finally {
+      em.transaction.commit()
+      em.close()
+    }
+  }
+
+  override suspend fun keys(): Iterable<K> {
+    val em = entityManagerProvider()
+    val query = em.createQuery(em.criteriaBuilder.createQuery(entityClass))
+    val resultStream: Stream<V> = query.resultStream
+    return Iterable { resultStream.map(idAccessor).iterator() }
+  }
+
+  override fun close() {
+  }
+}
diff --git a/kv/src/main/kotlin/org/apache/tuweni/kv/InfinispanKeyValueStore.kt b/kv/src/main/kotlin/org/apache/tuweni/kv/InfinispanKeyValueStore.kt
index 85a8b92..50e353f 100644
--- a/kv/src/main/kotlin/org/apache/tuweni/kv/InfinispanKeyValueStore.kt
+++ b/kv/src/main/kotlin/org/apache/tuweni/kv/InfinispanKeyValueStore.kt
@@ -18,7 +18,7 @@
 
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.future.await
-import org.apache.tuweni.bytes.Bytes
+import org.checkerframework.checker.units.qual.K
 import org.infinispan.Cache
 import kotlin.coroutines.CoroutineContext
 
@@ -26,10 +26,10 @@
  * A key-value store backed by [Infinispan](https://infinispan.org)
  *
  */
-class InfinispanKeyValueStore constructor(
-  private val cache: Cache<Bytes, Bytes>,
+class InfinispanKeyValueStore<K, V> constructor(
+  private val cache: Cache<K, V>,
   override val coroutineContext: CoroutineContext = Dispatchers.IO
-) : KeyValueStore {
+) : KeyValueStore<K, V> {
 
   companion object {
 
@@ -40,16 +40,16 @@
      * @return A key-value store.
      */
     @JvmStatic
-    fun open(cache: Cache<Bytes, Bytes>) = InfinispanKeyValueStore(cache)
+    fun <K, V> open(cache: Cache<K, V>) = InfinispanKeyValueStore(cache)
   }
 
-  override suspend fun get(key: Bytes): Bytes? = cache.getAsync(key).await()
+  override suspend fun get(key: K): V? = cache.getAsync(key).await()
 
-  override suspend fun put(key: Bytes, value: Bytes) {
+  override suspend fun put(key: K, value: V) {
     cache.putAsync(key, value).await()
   }
 
-  override suspend fun keys(): Iterable<Bytes> = cache.keys
+  override suspend fun keys(): Iterable<K> = cache.keys
 
   /**
    * The cache is managed outside the scope of this key-value store.
diff --git a/kv/src/main/kotlin/org/apache/tuweni/kv/KeyValueStore.kt b/kv/src/main/kotlin/org/apache/tuweni/kv/KeyValueStore.kt
index 588eddf..7f2775a 100644
--- a/kv/src/main/kotlin/org/apache/tuweni/kv/KeyValueStore.kt
+++ b/kv/src/main/kotlin/org/apache/tuweni/kv/KeyValueStore.kt
@@ -17,7 +17,6 @@
 package org.apache.tuweni.kv
 
 import kotlinx.coroutines.CoroutineScope
-import org.apache.tuweni.bytes.Bytes
 import org.apache.tuweni.concurrent.AsyncCompletion
 import org.apache.tuweni.concurrent.AsyncResult
 import org.apache.tuweni.concurrent.coroutines.asyncCompletion
@@ -27,7 +26,7 @@
 /**
  * A key-value store.
  */
-interface KeyValueStore : Closeable, CoroutineScope {
+interface KeyValueStore<K, V> : Closeable, CoroutineScope {
 
   /**
    * Retrieves data from the store.
@@ -35,7 +34,7 @@
    * @param key The key for the content.
    * @return The stored data, or null if no data was stored under the specified key.
    */
-  suspend fun get(key: Bytes): Bytes?
+  suspend fun get(key: K): V?
 
   /**
    * Retrieves data from the store.
@@ -44,7 +43,7 @@
    * @return An [AsyncResult] that will complete with the stored content,
    *         or an empty optional if no content was available.
    */
-  fun getAsync(key: Bytes): AsyncResult<Bytes?> = asyncResult { get(key) }
+  fun getAsync(key: K): AsyncResult<V?> = asyncResult { get(key) }
 
   /**
    * Puts data into the store.
@@ -52,7 +51,7 @@
    * @param key The key to associate with the data, for use when retrieving.
    * @param value The data to store.
    */
-  suspend fun put(key: Bytes, value: Bytes)
+  suspend fun put(key: K, value: V)
 
   /**
    * Puts data into the store.
@@ -64,19 +63,19 @@
    * @param value The data to store.
    * @return An [AsyncCompletion] that will complete when the content is stored.
    */
-  fun putAsync(key: Bytes, value: Bytes): AsyncCompletion = asyncCompletion { put(key, value) }
+  fun putAsync(key: K, value: V): AsyncCompletion = asyncCompletion { put(key, value) }
 
   /**
    * Provides an iterator over the keys of the store.
    *
    * @return An [Iterable] allowing to iterate over the set of keys.
    */
-  suspend fun keys(): Iterable<Bytes>
+  suspend fun keys(): Iterable<K>
 
   /**
    * Provides an iterator over the keys of the store.
    *
    * @return An [Iterable] allowing to iterate over the set of keys.
    */
-  fun keysAsync(): AsyncResult<Iterable<Bytes>> = asyncResult { keys() }
+  fun keysAsync(): AsyncResult<Iterable<K>> = asyncResult { keys() }
 }
diff --git a/kv/src/main/kotlin/org/apache/tuweni/kv/LevelDBKeyValueStore.kt b/kv/src/main/kotlin/org/apache/tuweni/kv/LevelDBKeyValueStore.kt
index 21ba81d..f0e8985 100644
--- a/kv/src/main/kotlin/org/apache/tuweni/kv/LevelDBKeyValueStore.kt
+++ b/kv/src/main/kotlin/org/apache/tuweni/kv/LevelDBKeyValueStore.kt
@@ -25,49 +25,91 @@
 import java.io.IOException
 import java.nio.file.Files
 import java.nio.file.Path
+import java.util.function.Function
 import kotlin.coroutines.CoroutineContext
 
 /**
  * A key-value store backed by LevelDB.
  *
  * @param dbPath The path to the levelDB database.
+ * @param keySerializer the serializer of key objects to bytes
+ * @param valueSerializer the serializer of value objects to bytes
+ * @param keyDeserializer the deserializer of keys from bytes
+ * @param valueDeserializer the deserializer of values from bytes
  * @param options Options for the levelDB database.
  * @param coroutineContext The co-routine context for blocking tasks.
  * @return A key-value store.
  * @throws IOException If an I/O error occurs.
  * @constructor Open a LevelDB-backed key-value store.
  */
-class LevelDBKeyValueStore
+class LevelDBKeyValueStore<K, V>
 @Throws(IOException::class)
 constructor(
   dbPath: Path,
+  private val keySerializer: (K) -> Bytes,
+  private val valueSerializer: (V) -> Bytes,
+  private val keyDeserializer: (Bytes) -> K,
+  private val valueDeserializer: (Bytes) -> V,
   options: Options = Options().createIfMissing(true).cacheSize((100 * 1048576).toLong()),
   override val coroutineContext: CoroutineContext = Dispatchers.IO
-) : KeyValueStore {
+) : KeyValueStore<K, V> {
 
   companion object {
+
     /**
      * Open a LevelDB-backed key-value store.
      *
      * @param dbPath The path to the levelDB database.
+     * @param keySerializer the serializer of key objects to bytes
+     * @param valueSerializer the serializer of value objects to bytes
+     * @param keyDeserializer the deserializer of keys from bytes
+     * @param valueDeserializer the deserializer of values from bytes
      * @return A key-value store.
      * @throws IOException If an I/O error occurs.
      */
     @JvmStatic
     @Throws(IOException::class)
-    fun open(dbPath: Path) = LevelDBKeyValueStore(dbPath)
+    fun <K, V> open(
+      dbPath: Path,
+      keySerializer: Function<K, Bytes>,
+      valueSerializer: Function<V, Bytes>,
+      keyDeserializer: Function<Bytes, K>,
+      valueDeserializer: Function<Bytes, V>
+    ): LevelDBKeyValueStore<K, V> =
+      LevelDBKeyValueStore(dbPath,
+        keySerializer::apply,
+        valueSerializer::apply,
+        keyDeserializer::apply,
+        valueDeserializer::apply)
 
     /**
      * Open a LevelDB-backed key-value store.
      *
      * @param dbPath The path to the levelDB database.
+     * @param keySerializer the serializer of key objects to bytes
+     * @param valueSerializer the serializer of value objects to bytes
+     * @param keyDeserializer the deserializer of keys from bytes
+     * @param valueDeserializer the deserializer of values from bytes
      * @param options Options for the levelDB database.
      * @return A key-value store.
      * @throws IOException If an I/O error occurs.
      */
     @JvmStatic
     @Throws(IOException::class)
-    fun open(dbPath: Path, options: Options) = LevelDBKeyValueStore(dbPath, options)
+    fun <K, V> open(
+      dbPath: Path,
+      keySerializer: Function<K, Bytes>,
+      valueSerializer: Function<V, Bytes>,
+      keyDeserializer: Function<Bytes, K>,
+      valueDeserializer: Function<Bytes, V>,
+      options: Options
+    ) =
+      LevelDBKeyValueStore(dbPath,
+        keySerializer::apply,
+        valueSerializer::apply,
+        keyDeserializer::apply,
+        valueDeserializer::apply,
+        options)
   }
 
   private val db: DB
@@ -77,27 +119,28 @@
     db = JniDBFactory.factory.open(dbPath.toFile(), options)
   }
 
-  override suspend fun get(key: Bytes): Bytes? {
-    val rawValue = db[key.toArrayUnsafe()]
+  override suspend fun get(key: K): V? {
+    val rawValue = db[keySerializer(key).toArrayUnsafe()]
     return if (rawValue == null) {
       null
     } else {
-      Bytes.wrap(rawValue)
+      valueDeserializer(Bytes.wrap(rawValue))
     }
   }
 
-  override suspend fun put(key: Bytes, value: Bytes) = db.put(key.toArrayUnsafe(), value.toArrayUnsafe())
+  override suspend fun put(key: K, value: V) = db.put(keySerializer(key).toArrayUnsafe(),
+    valueSerializer(value).toArrayUnsafe())
 
-  private class BytesIterator(val iter: DBIterator) : Iterator<Bytes> {
+  private class KIterator<K>(val iter: DBIterator, val keyDeserializer: (Bytes) -> K) : Iterator<K> {
     override fun hasNext(): Boolean = iter.hasNext()
 
-    override fun next(): Bytes = Bytes.wrap(iter.next().key)
+    override fun next(): K = keyDeserializer(Bytes.wrap(iter.next().key))
   }
 
-  override suspend fun keys(): Iterable<Bytes> {
+  override suspend fun keys(): Iterable<K> {
     val iter = db.iterator()
     iter.seekToFirst()
-    return Iterable { BytesIterator(iter) }
+    return Iterable { KIterator(iter, keyDeserializer) }
   }
 
   /**
diff --git a/kv/src/main/kotlin/org/apache/tuweni/kv/MapDBKeyValueStore.kt b/kv/src/main/kotlin/org/apache/tuweni/kv/MapDBKeyValueStore.kt
index f5bf128..190108f 100644
--- a/kv/src/main/kotlin/org/apache/tuweni/kv/MapDBKeyValueStore.kt
+++ b/kv/src/main/kotlin/org/apache/tuweni/kv/MapDBKeyValueStore.kt
@@ -32,29 +32,48 @@
  * A key-value store backed by a MapDB instance.
  *
  * @param dbPath The path to the MapDB database.
+ * @param keySerializer the serializer of key objects to bytes
+ * @param valueSerializer the serializer of value objects to bytes
+ * @param keyDeserializer the deserializer of keys from bytes
+ * @param valueDeserializer the deserializer of values from bytes
  * @param coroutineContext The co-routine context for blocking tasks.
  * @return A key-value store.
  * @throws IOException If an I/O error occurs.
  * @constructor Open a MapDB-backed key-value store.
  */
-class MapDBKeyValueStore
+class MapDBKeyValueStore<K, V>
 @Throws(IOException::class)
 constructor(
   dbPath: Path,
+  private val keySerializer: (K) -> Bytes,
+  private val valueSerializer: (V) -> Bytes,
+  private val keyDeserializer: (Bytes) -> K,
+  private val valueDeserializer: (Bytes?) -> V?,
   override val coroutineContext: CoroutineContext = Dispatchers.IO
-) : KeyValueStore {
+) : KeyValueStore<K, V> {
 
   companion object {
     /**
      * Open a MapDB-backed key-value store.
      *
      * @param dbPath The path to the MapDB database.
+     * @param keySerializer the serializer of key objects to bytes
+     * @param valueSerializer the serializer of value objects to bytes
+     * @param keyDeserializer the deserializer of keys from bytes
+     * @param valueDeserializer the deserializer of values from bytes
      * @return A key-value store.
      * @throws IOException If an I/O error occurs.
      */
     @JvmStatic
     @Throws(IOException::class)
-    fun open(dbPath: Path) = MapDBKeyValueStore(dbPath)
+    fun <K, V> open(
+      dbPath: Path,
+      keySerializer: (K) -> Bytes,
+      valueSerializer: (V) -> Bytes,
+      keyDeserializer: (Bytes) -> K,
+      valueDeserializer: (Bytes?) -> V?
+    ) =
+      MapDBKeyValueStore<K, V>(dbPath, keySerializer, valueSerializer, keyDeserializer, valueDeserializer)
   }
 
   private val db: DB
@@ -70,14 +89,14 @@
     ).createOrOpen()
   }
 
-  override suspend fun get(key: Bytes): Bytes? = storageData[key]
+  override suspend fun get(key: K): V? = valueDeserializer(storageData[keySerializer(key)])
 
-  override suspend fun put(key: Bytes, value: Bytes) {
-    storageData[key] = value
+  override suspend fun put(key: K, value: V) {
+    storageData[keySerializer(key)] = valueSerializer(value)
     db.commit()
   }
 
-  override suspend fun keys(): Iterable<Bytes> = storageData.keys
+  override suspend fun keys(): Iterable<K> = storageData.keys.map(keyDeserializer)
 
   /**
    * Closes the underlying MapDB instance.
diff --git a/kv/src/main/kotlin/org/apache/tuweni/kv/MapKeyValueStore.kt b/kv/src/main/kotlin/org/apache/tuweni/kv/MapKeyValueStore.kt
index bc15460..5eaca56 100644
--- a/kv/src/main/kotlin/org/apache/tuweni/kv/MapKeyValueStore.kt
+++ b/kv/src/main/kotlin/org/apache/tuweni/kv/MapKeyValueStore.kt
@@ -17,7 +17,6 @@
 package org.apache.tuweni.kv
 
 import kotlinx.coroutines.Dispatchers
-import org.apache.tuweni.bytes.Bytes
 import kotlin.coroutines.CoroutineContext
 
 /**
@@ -27,11 +26,11 @@
  * @return A key-value store.
  * @constructor Open an in-memory key-value store.
  */
-class MapKeyValueStore
+class MapKeyValueStore<K, V>
 constructor(
-  private val map: MutableMap<Bytes, Bytes> = HashMap(),
+  private val map: MutableMap<K, V> = HashMap(),
   override val coroutineContext: CoroutineContext = Dispatchers.IO
-) : KeyValueStore {
+) : KeyValueStore<K, V> {
 
   companion object {
     /**
@@ -42,7 +41,7 @@
      * @return A key-value store.
      */
     @JvmStatic
-    fun open(): MapKeyValueStore = MapKeyValueStore()
+    fun <K, V> open() = MapKeyValueStore<K, V>()
 
     /**
      * Open an in-memory key-value store.
@@ -51,16 +50,16 @@
      * @return A key-value store.
      */
     @JvmStatic
-    fun open(map: MutableMap<Bytes, Bytes>) = MapKeyValueStore(map)
+    fun <K, V> open(map: MutableMap<K, V>) = MapKeyValueStore(map)
   }
 
-  override suspend fun get(key: Bytes): Bytes? = map[key]
+  override suspend fun get(key: K): V? = map[key]
 
-  override suspend fun put(key: Bytes, value: Bytes) {
+  override suspend fun put(key: K, value: V) {
     map[key] = value
   }
 
-  override suspend fun keys(): Iterable<Bytes> = map.keys
+  override suspend fun keys(): Iterable<K> = map.keys
 
   /**
    * Has no effect in this KeyValueStore implementation.
diff --git a/kv/src/main/kotlin/org/apache/tuweni/kv/RedisKeyValueStore.kt b/kv/src/main/kotlin/org/apache/tuweni/kv/RedisKeyValueStore.kt
index 8e8717c..f22f0cd 100644
--- a/kv/src/main/kotlin/org/apache/tuweni/kv/RedisKeyValueStore.kt
+++ b/kv/src/main/kotlin/org/apache/tuweni/kv/RedisKeyValueStore.kt
@@ -24,58 +24,131 @@
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.future.await
 import org.apache.tuweni.bytes.Bytes
+import org.checkerframework.checker.units.qual.K
 import java.net.InetAddress
 import java.util.concurrent.CompletionStage
+import java.util.function.Function
 import kotlin.coroutines.CoroutineContext
 
 /**
  * A key-value store backed by Redis.
  *
  * @param uri The uri to the Redis store.
+ * @param keySerializer the serializer of key objects to bytes
+ * @param valueSerializer the serializer of value objects to bytes
+ * @param keyDeserializer the deserializer of keys from bytes
+ * @param valueDeserializer the deserializer of values from bytes
+ * @param coroutineContext the co-routine context in which this store executes
  * @constructor Open a Redis-backed key-value store.
  */
-class RedisKeyValueStore(
+class RedisKeyValueStore<K, V>(
   uri: String,
+  private val keySerializer: (K) -> Bytes,
+  private val valueSerializer: (V) -> Bytes,
+  private val keyDeserializer: (Bytes) -> K,
+  private val valueDeserializer: (Bytes) -> V,
   override val coroutineContext: CoroutineContext = Dispatchers.IO
-) : KeyValueStore {
+) : KeyValueStore<K, V> {
 
   companion object {
     /**
      * Open a Redis-backed key-value store.
      *
      * @param uri The uri to the Redis store.
+     * @param keySerializer the serializer of key objects to bytes
+     * @param valueSerializer the serializer of value objects to bytes
+     * @param keyDeserializer the deserializer of keys from bytes
+     * @param valueDeserializer the deserializer of values from bytes
      * @return A key-value store.
      */
     @JvmStatic
-    fun open(uri: String) = RedisKeyValueStore(uri)
+    fun <K, V> open(
+      uri: String,
+      keySerializer: Function<K, Bytes>,
+      valueSerializer: Function<V, Bytes>,
+      keyDeserializer: Function<Bytes, K>,
+      valueDeserializer: Function<Bytes, V>
+    ) = RedisKeyValueStore(uri,
+      keySerializer::apply,
+      valueSerializer::apply,
+      keyDeserializer::apply,
+      valueDeserializer::apply)
 
     /**
      * Open a Redis-backed key-value store.
      *
      * @param port The port for the Redis store.
+     * @param keySerializer the serializer of key objects to bytes
+     * @param valueSerializer the serializer of value objects to bytes
+     * @param keyDeserializer the deserializer of keys from bytes
+     * @param valueDeserializer the deserializer of values from bytes
      * @return A key-value store.
      */
     @JvmStatic
-    fun open(port: Int) = RedisKeyValueStore(port)
+    fun <K, V> open(
+      port: Int,
+      keySerializer: Function<K, Bytes>,
+      valueSerializer: Function<V, Bytes>,
+      keyDeserializer: Function<Bytes, K>,
+      valueDeserializer: Function<Bytes, V>
+    ) =
+      RedisKeyValueStore(port = port,
+        keySerializer = keySerializer::apply,
+        valueSerializer = valueSerializer::apply,
+        keyDeserializer = keyDeserializer::apply,
+        valueDeserializer = valueDeserializer::apply)
 
     /**
      * Open a Redis-backed key-value store.
      *
      * @param address The address for the Redis store.
+     * @param keySerializer the serializer of key objects to bytes
+     * @param valueSerializer the serializer of value objects to bytes
+     * @param keyDeserializer the deserializer of keys from bytes
+     * @param valueDeserializer the deserializer of values from bytes
      * @return A key-value store.
      */
     @JvmStatic
-    fun open(address: InetAddress) = RedisKeyValueStore(6379, address)
+    fun <K, V> open(
+      address: InetAddress,
+      keySerializer: Function<K, Bytes>,
+      valueSerializer: Function<V, Bytes>,
+      keyDeserializer: Function<Bytes, K>,
+      valueDeserializer: Function<Bytes, V>
+    ) =
+      RedisKeyValueStore(6379,
+        address,
+        keySerializer::apply,
+        valueSerializer::apply,
+        keyDeserializer::apply,
+        valueDeserializer::apply)
 
     /**
      * Open a Redis-backed key-value store.
      *
      * @param port The port for the Redis store.
      * @param address The address for the Redis store.
+     * @param keySerializer the serializer of key objects to bytes
+     * @param valueSerializer the serializer of value objects to bytes
+     * @param keyDeserializer the deserializer of keys from bytes
+     * @param valueDeserializer the deserializer of values from bytes
      * @return A key-value store.
      */
     @JvmStatic
-    fun open(port: Int, address: InetAddress) = RedisKeyValueStore(port, address)
+    fun <K, V> open(
+      port: Int,
+      address: InetAddress,
+      keySerializer: Function<K, Bytes>,
+      valueSerializer: Function<V, Bytes>,
+      keyDeserializer: Function<Bytes, K>,
+      valueDeserializer: Function<Bytes, V>
+    ) =
+      RedisKeyValueStore(port,
+        address,
+        keySerializer::apply,
+        valueSerializer::apply,
+        keyDeserializer::apply,
+        valueDeserializer::apply)
 
     /**
      * A [RedisCodec] for working with Bytes classes.
@@ -94,12 +167,24 @@
    *
    * @param port The port for the Redis store.
    * @param address The address for the Redis store.
+   * @param keySerializer the serializer of key objects to bytes
+   * @param valueSerializer the serializer of value objects to bytes
+   * @param keyDeserializer the deserializer of keys from bytes
+   * @param valueDeserializer the deserializer of values from bytes
    */
   @JvmOverloads
   constructor(
     port: Int = 6379,
-    address: InetAddress = InetAddress.getLoopbackAddress()
-  ) : this(RedisURI.create(address.hostAddress, port).toURI().toString())
+    address: InetAddress = InetAddress.getLoopbackAddress(),
+    keySerializer: (K) -> Bytes,
+    valueSerializer: (V) -> Bytes,
+    keyDeserializer: (Bytes) -> K,
+    valueDeserializer: (Bytes) -> V
+  ) : this(RedisURI.create(address.hostAddress, port).toURI().toString(),
+    keySerializer,
+    valueSerializer,
+    keyDeserializer,
+    valueDeserializer)
 
   init {
     val redisClient = RedisClient.create(uri)
@@ -107,14 +192,20 @@
     asyncCommands = conn.async()
   }
 
-  override suspend fun get(key: Bytes): Bytes? = asyncCommands.get(key).await()
+  override suspend fun get(key: K): V? = asyncCommands.get(keySerializer(key)).thenApply {
+    if (it == null) {
+      null
+    } else {
+      valueDeserializer(it)
+    }
+  }.await()
 
-  override suspend fun put(key: Bytes, value: Bytes) {
-    val future: CompletionStage<String> = asyncCommands.set(key, value)
+  override suspend fun put(key: K, value: V) {
+    val future: CompletionStage<String> = asyncCommands.set(keySerializer(key), valueSerializer(value))
     future.await()
   }
 
-  override suspend fun keys(): Iterable<Bytes> = asyncCommands.keys(Bytes.EMPTY).await()
+  override suspend fun keys(): Iterable<K> = asyncCommands.keys(Bytes.EMPTY).await().map(keyDeserializer)
 
   override fun close() {
     conn.close()
diff --git a/kv/src/main/kotlin/org/apache/tuweni/kv/RocksDBKeyValueStore.kt b/kv/src/main/kotlin/org/apache/tuweni/kv/RocksDBKeyValueStore.kt
index 7911803..30e8dad 100644
--- a/kv/src/main/kotlin/org/apache/tuweni/kv/RocksDBKeyValueStore.kt
+++ b/kv/src/main/kotlin/org/apache/tuweni/kv/RocksDBKeyValueStore.kt
@@ -25,6 +25,7 @@
 import java.nio.file.Files
 import java.nio.file.Path
 import java.util.concurrent.atomic.AtomicBoolean
+import java.util.function.Function
 import kotlin.coroutines.CoroutineContext
 
 /**
@@ -37,37 +38,72 @@
  * @throws IOException If an I/O error occurs.
  * @constructor Open a RocksDB-backed key-value store.
  */
-class RocksDBKeyValueStore
+class RocksDBKeyValueStore<K, V>
 @Throws(IOException::class)
 constructor(
   dbPath: Path,
+  private val keySerializer: (K) -> Bytes,
+  private val valueSerializer: (V) -> Bytes,
+  private val keyDeserializer: (Bytes) -> K,
+  private val valueDeserializer: (Bytes) -> V,
   options: Options = Options().setCreateIfMissing(true).setWriteBufferSize(268435456).setMaxOpenFiles(-1),
   override val coroutineContext: CoroutineContext = Dispatchers.IO
-) : KeyValueStore {
+) : KeyValueStore<K, V> {
 
   companion object {
     /**
      * Open a RocksDB-backed key-value store.
      *
      * @param dbPath The path to the RocksDB database.
+     * @param keySerializer the serializer of key objects to bytes
+     * @param valueSerializer the serializer of value objects to bytes
+     * @param keyDeserializer the deserializer of keys from bytes
+     * @param valueDeserializer the deserializer of values from bytes
      * @return A key-value store.
      * @throws IOException If an I/O error occurs.
      */
     @JvmStatic
     @Throws(IOException::class)
-    fun open(dbPath: Path) = RocksDBKeyValueStore(dbPath)
+    fun <K, V> open(
+      dbPath: Path,
+      keySerializer: Function<K, Bytes>,
+      valueSerializer: Function<V, Bytes>,
+      keyDeserializer: Function<Bytes, K>,
+      valueDeserializer: Function<Bytes, V>
+    ) = RocksDBKeyValueStore(dbPath,
+      keySerializer::apply,
+      valueSerializer::apply,
+      keyDeserializer::apply,
+      valueDeserializer::apply)
 
     /**
      * Open a RocksDB-backed key-value store.
      *
      * @param dbPath The path to the RocksDB database.
+     * @param keySerializer the serializer of key objects to bytes
+     * @param valueSerializer the serializer of value objects to bytes
+     * @param keyDeserializer the deserializer of keys from bytes
+     * @param valueDeserializer the deserializer of values from bytes
      * @param options Options for the RocksDB database.
      * @return A key-value store.
      * @throws IOException If an I/O error occurs.
      */
     @JvmStatic
     @Throws(IOException::class)
-    fun open(dbPath: Path, options: Options) = RocksDBKeyValueStore(dbPath, options)
+    fun <K, V> open(
+      dbPath: Path,
+      keySerializer: Function<K, Bytes>,
+      valueSerializer: Function<V, Bytes>,
+      keyDeserializer: Function<Bytes, K>,
+      valueDeserializer: Function<Bytes, V>,
+      options: Options
+    ) =
+      RocksDBKeyValueStore(dbPath,
+        keySerializer::apply,
+        valueSerializer::apply,
+        keyDeserializer::apply,
+        valueDeserializer::apply,
+        options)
   }
 
   private val db: RocksDB
@@ -79,43 +115,43 @@
     db = RocksDB.open(options, dbPath.toAbsolutePath().toString())
   }
 
-  override suspend fun get(key: Bytes): Bytes? {
+  override suspend fun get(key: K): V? {
     if (closed.get()) {
       throw IllegalStateException("Closed DB")
     }
-    val rawValue = db[key.toArrayUnsafe()]
+    val rawValue = db[keySerializer(key).toArrayUnsafe()]
     return if (rawValue == null) {
       null
     } else {
-      Bytes.wrap(rawValue)
+      valueDeserializer(Bytes.wrap(rawValue))
     }
   }
 
-  override suspend fun put(key: Bytes, value: Bytes) {
+  override suspend fun put(key: K, value: V) {
     if (closed.get()) {
       throw IllegalStateException("Closed DB")
     }
-    db.put(key.toArrayUnsafe(), value.toArrayUnsafe())
+    db.put(keySerializer(key).toArrayUnsafe(), valueSerializer(value).toArrayUnsafe())
   }
 
-  private class BytesIterator(val rIterator: RocksIterator) : Iterator<Bytes> {
+  private class BytesIterator<K>(val rIterator: RocksIterator, val keyDeserializer: (Bytes) -> K) : Iterator<K> {
 
     override fun hasNext(): Boolean = rIterator.isValid
 
-    override fun next(): Bytes {
-      val key = Bytes.wrap(rIterator.key())
+    override fun next(): K {
+      val key = rIterator.key()
       rIterator.next()
-      return key
+      return keyDeserializer(Bytes.wrap(key))
     }
   }
 
-  override suspend fun keys(): Iterable<Bytes> {
+  override suspend fun keys(): Iterable<K> {
     if (closed.get()) {
       throw IllegalStateException("Closed DB")
     }
     val iter = db.newIterator()
     iter.seekToFirst()
-    return Iterable { BytesIterator(iter) }
+    return Iterable { BytesIterator(iter, keyDeserializer) }
   }
 
   /**
diff --git a/kv/src/main/kotlin/org/apache/tuweni/kv/SQLKeyValueStore.kt b/kv/src/main/kotlin/org/apache/tuweni/kv/SQLKeyValueStore.kt
index aa07336..46c6e93 100644
--- a/kv/src/main/kotlin/org/apache/tuweni/kv/SQLKeyValueStore.kt
+++ b/kv/src/main/kotlin/org/apache/tuweni/kv/SQLKeyValueStore.kt
@@ -38,27 +38,45 @@
  * @throws IOException If an I/O error occurs.
  * @constructor Open a relational database backed key-value store.
  */
-class SQLKeyValueStore
+class SQLKeyValueStore<K, V>
 @Throws(IOException::class)
 constructor(
   jdbcurl: String,
   val tableName: String = "store",
   val keyColumn: String = "key",
   val valueColumn: String = "value",
+  private val keySerializer: (K) -> Bytes,
+  private val valueSerializer: (V) -> Bytes,
+  private val keyDeserializer: (Bytes) -> K,
+  private val valueDeserializer: (Bytes?) -> V?,
   override val coroutineContext: CoroutineContext = Dispatchers.IO
-) : KeyValueStore {
+) : KeyValueStore<K, V> {
 
   companion object {
     /**
      * Open a relational database backed key-value store.
      *
      * @param jdbcUrl The JDBC url to connect to the database.
+     * @param keySerializer the serializer of key objects to bytes
+     * @param valueSerializer the serializer of value objects to bytes
+     * @param keyDeserializer the deserializer of keys from bytes
+     * @param valueDeserializer the deserializer of values from bytes
      * @return A key-value store.
      * @throws IOException If an I/O error occurs.
      */
     @JvmStatic
     @Throws(IOException::class)
-    fun open(jdbcUrl: String) = SQLKeyValueStore(jdbcUrl)
+    fun <K, V> open(
+      jdbcUrl: String,
+      keySerializer: (K) -> Bytes,
+      valueSerializer: (V) -> Bytes,
+      keyDeserializer: (Bytes) -> K,
+      valueDeserializer: (Bytes?) -> V?
+    ) = SQLKeyValueStore<K, V>(jdbcurl = jdbcUrl,
+      keySerializer = keySerializer,
+      valueSerializer = valueSerializer,
+    keyDeserializer = keyDeserializer,
+      valueDeserializer = valueDeserializer)
 
     /**
      * Open a relational database backed key-value store.
@@ -67,13 +85,33 @@
      * @param tableName the name of the table to use for storage.
      * @param keyColumn the key column of the store.
      * @param valueColumn the value column of the store.
+     * @param keySerializer the serializer of key objects to bytes
+     * @param valueSerializer the serializer of value objects to bytes
+     * @param keyDeserializer the deserializer of keys from bytes
+     * @param valueDeserializer the deserializer of values from bytes
      * @return A key-value store.
      * @throws IOException If an I/O error occurs.
      */
     @JvmStatic
     @Throws(IOException::class)
-    fun open(jdbcUrl: String, tableName: String, keyColumn: String, valueColumn: String) =
-      SQLKeyValueStore(jdbcUrl, tableName, keyColumn, valueColumn)
+    fun <K, V> open(
+      jdbcUrl: String,
+      tableName: String,
+      keyColumn: String,
+      valueColumn: String,
+      keySerializer: (K) -> Bytes,
+      valueSerializer: (V) -> Bytes,
+      keyDeserializer: (Bytes) -> K,
+      valueDeserializer: (Bytes?) -> V?
+    ) =
+      SQLKeyValueStore<K, V>(jdbcUrl,
+        tableName,
+        keyColumn,
+        valueColumn,
+        keySerializer,
+        valueSerializer,
+        keyDeserializer,
+        valueDeserializer)
   }
 
   private val connectionPool: BoneCP
@@ -85,34 +123,34 @@
     connectionPool = BoneCP(config)
   }
 
-  override suspend fun get(key: Bytes): Bytes? {
+  override suspend fun get(key: K): V? {
       connectionPool.asyncConnection.await().use {
         val stmt = it.prepareStatement("SELECT $valueColumn FROM $tableName WHERE $keyColumn = ?")
-        stmt.setBytes(1, key.toArrayUnsafe())
+        stmt.setBytes(1, keySerializer(key).toArrayUnsafe())
         stmt.execute()
 
         val rs = stmt.resultSet
 
         return if (rs.next()) {
-          Bytes.wrap(rs.getBytes(1))
+          valueDeserializer(Bytes.wrap(rs.getBytes(1)))
         } else {
           null
         }
       }
   }
 
-  override suspend fun put(key: Bytes, value: Bytes) {
+  override suspend fun put(key: K, value: V) {
     connectionPool.asyncConnection.await().use {
         val stmt = it.prepareStatement("INSERT INTO $tableName($keyColumn, $valueColumn) VALUES(?,?)")
-        stmt.setBytes(1, key.toArrayUnsafe())
-        stmt.setBytes(2, value.toArrayUnsafe())
+        stmt.setBytes(1, keySerializer(key).toArrayUnsafe())
+        stmt.setBytes(2, valueSerializer(value).toArrayUnsafe())
         it.autoCommit = false
         try {
           stmt.execute()
         } catch (e: SQLException) {
           val updateStmt = it.prepareStatement("UPDATE $tableName SET $valueColumn=? WHERE $keyColumn=?")
-          updateStmt.setBytes(1, value.toArrayUnsafe())
-          updateStmt.setBytes(2, key.toArrayUnsafe())
+          updateStmt.setBytes(1, valueSerializer(value).toArrayUnsafe())
+          updateStmt.setBytes(2, keySerializer(key).toArrayUnsafe())
           updateStmt.execute()
         }
         it.commit()
@@ -120,24 +158,24 @@
       }
   }
 
-  private class SQLIterator(val resultSet: ResultSet) : Iterator<Bytes> {
+  private class SQLIterator<K>(val resultSet: ResultSet, val keyDeserializer: (Bytes) -> K) : Iterator<K> {
 
     private var next = resultSet.next()
 
     override fun hasNext(): Boolean = next
 
-    override fun next(): Bytes {
-      val key = Bytes.wrap(resultSet.getBytes(1))
+    override fun next(): K {
+      val key = keyDeserializer(Bytes.wrap(resultSet.getBytes(1)))
       next = resultSet.next()
       return key
     }
   }
 
-  override suspend fun keys(): Iterable<Bytes> {
+  override suspend fun keys(): Iterable<K> {
     connectionPool.asyncConnection.await().use {
       val stmt = it.prepareStatement("SELECT $keyColumn FROM $tableName")
       stmt.execute()
-      return Iterable { SQLIterator(stmt.resultSet) }
+      return Iterable { SQLIterator(stmt.resultSet, keyDeserializer) }
     }
   }
 
diff --git a/kv/src/test/java/org/apache/tuweni/kv/KeyValueStoreTest.java b/kv/src/test/java/org/apache/tuweni/kv/KeyValueStoreTest.java
index b5d8c2a..19e0f37 100644
--- a/kv/src/test/java/org/apache/tuweni/kv/KeyValueStoreTest.java
+++ b/kv/src/test/java/org/apache/tuweni/kv/KeyValueStoreTest.java
@@ -27,6 +27,7 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.function.Function;
 
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -34,10 +35,12 @@
 @ExtendWith(TempDirectoryExtension.class)
 class KeyValueStoreTest {
 
+  private static final Function<Bytes, Bytes> bytesIdentityFn = Function.identity();
+
   @Test
   void testPutAndGet() throws Exception {
     Map<Bytes, Bytes> map = new HashMap<>();
-    KeyValueStore store = MapKeyValueStore.open(map);
+    KeyValueStore<Bytes, Bytes> store = MapKeyValueStore.open(map);
     AsyncCompletion completion = store.putAsync(Bytes.of(123), Bytes.of(10, 12, 13));
     completion.join();
     Bytes value = store.getAsync(Bytes.of(123)).get();
@@ -49,14 +52,14 @@
   @Test
   void testNoValue() throws Exception {
     Map<Bytes, Bytes> map = new HashMap<>();
-    KeyValueStore store = MapKeyValueStore.open(map);
+    KeyValueStore<Bytes, Bytes> store = MapKeyValueStore.open(map);
     assertNull(store.getAsync(Bytes.of(123)).get());
   }
 
   @Test
   void testKeys() throws Exception {
     Map<Bytes, Bytes> map = new HashMap<>();
-    KeyValueStore store = MapKeyValueStore.open(map);
+    KeyValueStore<Bytes, Bytes> store = MapKeyValueStore.open(map);
     AsyncCompletion completion = store.putAsync(Bytes.of(123), Bytes.of(10, 12, 13));
     completion.join();
     Set<Bytes> keys = new HashSet<>();
@@ -66,7 +69,12 @@
 
   @Test
   void testLevelDBWithoutOptions(@TempDirectory Path tempDirectory) throws Exception {
-    try (LevelDBKeyValueStore leveldb = LevelDBKeyValueStore.open(tempDirectory.resolve("foo").resolve("bar"))) {
+    try (LevelDBKeyValueStore<Bytes, Bytes> leveldb = LevelDBKeyValueStore.open(
+        tempDirectory.resolve("foo").resolve("bar"),
+        bytesIdentityFn,
+        bytesIdentityFn,
+        bytesIdentityFn,
+        bytesIdentityFn)) {
       AsyncCompletion completion = leveldb.putAsync(Bytes.of(123), Bytes.of(10, 12, 13));
       completion.join();
       Bytes value = leveldb.getAsync(Bytes.of(123)).get();
@@ -77,7 +85,12 @@
 
   @Test
   void testRocksDBWithoutOptions(@TempDirectory Path tempDirectory) throws Exception {
-    try (RocksDBKeyValueStore rocksdb = RocksDBKeyValueStore.open(tempDirectory.resolve("foo").resolve("bar"))) {
+    try (RocksDBKeyValueStore<Bytes, Bytes> rocksdb = RocksDBKeyValueStore.open(
+        tempDirectory.resolve("foo").resolve("bar"),
+        bytesIdentityFn,
+        bytesIdentityFn,
+        bytesIdentityFn,
+        bytesIdentityFn)) {
       AsyncCompletion completion = rocksdb.putAsync(Bytes.of(123), Bytes.of(10, 12, 13));
       completion.join();
       Bytes value = rocksdb.getAsync(Bytes.of(123)).get();
diff --git a/kv/src/test/java/org/apache/tuweni/kv/RedisKeyValueStoreTest.java b/kv/src/test/java/org/apache/tuweni/kv/RedisKeyValueStoreTest.java
index c218a65..db44ae1 100644
--- a/kv/src/test/java/org/apache/tuweni/kv/RedisKeyValueStoreTest.java
+++ b/kv/src/test/java/org/apache/tuweni/kv/RedisKeyValueStoreTest.java
@@ -22,6 +22,7 @@
 import org.apache.tuweni.junit.RedisServerExtension;
 
 import java.net.InetAddress;
+import java.util.function.Function;
 
 import io.lettuce.core.RedisClient;
 import io.lettuce.core.RedisURI;
@@ -34,7 +35,8 @@
 
   @Test
   void testPutAndGet(@RedisPort Integer redisPort) throws Exception {
-    KeyValueStore store = RedisKeyValueStore.open(redisPort);
+    KeyValueStore<Bytes, Bytes> store = RedisKeyValueStore
+        .open(redisPort, Function.identity(), Function.identity(), Function.identity(), Function.identity());
     AsyncCompletion completion = store.putAsync(Bytes.of(123), Bytes.of(10, 12, 13));
     completion.join();
     Bytes value = store.getAsync(Bytes.of(123)).get();
@@ -49,13 +51,24 @@
 
   @Test
   void testNoValue(@RedisPort Integer redisPort) throws Exception {
-    KeyValueStore store = RedisKeyValueStore.open(redisPort, InetAddress.getLoopbackAddress());
+    KeyValueStore<Bytes, Bytes> store = RedisKeyValueStore.open(
+        redisPort,
+        InetAddress.getLoopbackAddress(),
+        Function.identity(),
+        Function.identity(),
+        Function.identity(),
+        Function.identity());
     assertNull(store.getAsync(Bytes.of(124)).get());
   }
 
   @Test
   void testRedisCloseable(@RedisPort Integer redisPort) throws Exception {
-    try (RedisKeyValueStore redis = RedisKeyValueStore.open("redis://127.0.0.1:" + redisPort)) {
+    try (RedisKeyValueStore<Bytes, Bytes> redis = RedisKeyValueStore.open(
+        "redis://127.0.0.1:" + redisPort,
+        Function.identity(),
+        Function.identity(),
+        Function.identity(),
+        Function.identity())) {
       AsyncCompletion completion = redis.putAsync(Bytes.of(125), Bytes.of(10, 12, 13));
       completion.join();
       Bytes value = redis.getAsync(Bytes.of(125)).get();
diff --git a/kv/src/test/kotlin/org/apache/tuweni/kv/KeyValueStoreSpec.kt b/kv/src/test/kotlin/org/apache/tuweni/kv/KeyValueStoreSpec.kt
index ae8c62b..94f4ffb 100644
--- a/kv/src/test/kotlin/org/apache/tuweni/kv/KeyValueStoreSpec.kt
+++ b/kv/src/test/kotlin/org/apache/tuweni/kv/KeyValueStoreSpec.kt
@@ -110,7 +110,7 @@
       runBlocking {
         kv.put(bar, foo)
         kv.put(foo, bar)
-          val keys = (kv.keys().map { it })
+        val keys = (kv.keys().map { it })
         keys.should.contain(bar)
         keys.should.contain(foo)
       }
@@ -120,7 +120,12 @@
 
 object MapDBKeyValueStoreSpec : Spek({
   val testDir = Files.createTempDirectory("data")
-  val kv = MapDBKeyValueStore(testDir.resolve("data.db"))
+  val kv = MapDBKeyValueStore(
+    testDir.resolve("data.db"),
+    keySerializer = { it },
+    valueSerializer = { it },
+    keyDeserializer = { it },
+    valueDeserializer = { it })
 
   describe("a MapDB-backed key value store") {
 
@@ -141,12 +146,19 @@
       runBlocking {
         kv.put(foobar, foo)
         kv.put(foo, bar)
-        kv.keys().should.equal(setOf(foobar, foo))
+        val keys = kv.keys().map { it }
+        keys.should.contain(foo)
+        keys.should.contain(foobar)
       }
     }
 
     it("should not allow usage after the DB is closed") {
-      val kv2 = MapDBKeyValueStore(testDir.resolve("data2.db"))
+      val kv2 = MapDBKeyValueStore(
+        testDir.resolve("data2.db"),
+        keySerializer = { it },
+        valueSerializer = { it },
+        keyDeserializer = { it },
+        valueDeserializer = { it })
       kv2.close()
       runBlocking {
         var caught = false
@@ -168,7 +180,12 @@
 
 object LevelDBKeyValueStoreSpec : Spek({
   val path = Files.createTempDirectory("leveldb")
-  val kv = LevelDBKeyValueStore(path)
+  val kv = LevelDBKeyValueStore(
+    path,
+    keySerializer = { it },
+    valueSerializer = { it },
+    keyDeserializer = { it },
+    valueDeserializer = { it })
   afterGroup {
     kv.close()
     MoreFiles.deleteRecursively(path, RecursiveDeleteOption.ALLOW_INSECURE)
@@ -197,7 +214,12 @@
     }
 
     it("should not allow usage after the DB is closed") {
-      val kv2 = LevelDBKeyValueStore(path.resolve("subdb"))
+      val kv2 = LevelDBKeyValueStore(
+        path.resolve("subdb"),
+        keySerializer = { it },
+        valueSerializer = { it },
+        keyDeserializer = { it },
+        valueDeserializer = { it })
       kv2.close()
       runBlocking {
         var caught = false
@@ -214,7 +236,12 @@
 
 object RocksDBKeyValueStoreSpec : Spek({
   val path = Files.createTempDirectory("rocksdb")
-  val kv = RocksDBKeyValueStore(path)
+  val kv = RocksDBKeyValueStore(
+    path,
+    keySerializer = { it },
+    valueSerializer = { it },
+    keyDeserializer = { it },
+    valueDeserializer = { it })
   afterGroup {
     kv.close()
     MoreFiles.deleteRecursively(path, RecursiveDeleteOption.ALLOW_INSECURE)
@@ -245,7 +272,12 @@
     }
 
     it("should not allow usage after the DB is closed") {
-      val kv2 = RocksDBKeyValueStore(path.resolve("subdb"))
+      val kv2 = RocksDBKeyValueStore(
+        path.resolve("subdb"),
+        keySerializer = { it },
+        valueSerializer = { it },
+        keyDeserializer = { it },
+        valueDeserializer = { it })
       kv2.close()
       runBlocking {
         var caught = false
@@ -269,8 +301,21 @@
     st.executeUpdate("create table store(key binary, value binary, primary key(key))")
     st.executeUpdate("create table store2(id binary, val binary, primary key(id))")
   }
-  val kv = SQLKeyValueStore(jdbcUrl)
-  val otherkv = SQLKeyValueStore.open(jdbcUrl, "store2", "id", "val")
+  val kv = SQLKeyValueStore(
+    jdbcUrl,
+    keySerializer = { it },
+    valueSerializer = { it },
+    keyDeserializer = { it },
+    valueDeserializer = { it })
+  val otherkv = SQLKeyValueStore.open(
+    jdbcUrl,
+    "store2",
+    "id",
+    "val",
+    keySerializer = { it },
+    valueSerializer = { it },
+    keyDeserializer = { it },
+    valueDeserializer = { it })
   afterGroup {
     kv.close()
     otherkv.close()
@@ -316,7 +361,12 @@
     }
 
     it("should not allow usage after the DB is closed") {
-      val kv2 = SQLKeyValueStore("jdbc:h2:mem:testdb")
+      val kv2 = SQLKeyValueStore(
+        "jdbc:h2:mem:testdb",
+        keySerializer = { it },
+        valueSerializer = { it },
+        keyDeserializer = { it },
+        valueDeserializer = { it })
       kv2.close()
       runBlocking {
         var caught = false