KMongo и неструктурированные данные

Несмотря на то, что MongoDB начало движение в сторону строгости реляционной модели (добавление опциональной схемы данных, join‑запросов, агрегаций и т. п.), она по прежнему остается документной базой данных и предполагает возможность сохранения документов произвольной структуры. И при использовании MongoDB в языках с динамической типизацией (JavaScript, Python) сохранение или генерация объектов не вызывает сложностей, поскольку заранее не требуется определить структуру извлекаемого или сохраняемого объекта. Но как действовать в случае использования драйверов для MongoDB для языков со строгой типизацией — например, библиотеки KMongo для Kotlin?

В этой статье мы разберем приемы для работы с неструктурированными данными, которые позволят сохранить преимущества использования сериализации (Jackson или kotlinx.serializable) с механизмами рефлексии для извлечения произвольных документов.

Для начала несколько слов о самой библиотеке KMongo. Open Source библиотека основана на Core MongoDB Java Driver и предоставляет удобный синтаксис для выполнения запросов и итерации по базам данных, коллекциям и результатам запроса с использованием возможностей Kotlin (в частности используется определение инфиксных операторов для создания запросов вида Jedi::age lt yoda?.age), а также обеспечивает поддержку неявной сериализации при чтении документов из Mongo и сохранения новых документов. Также KMongo предоставляет возможность регистрации адаптеров для использования механизмов неблокирующего выполнения операций (с использованием Rx или корутин).

Для установки KMongo необходимо добавить основную библиотеку и адаптер для неблокирующего выполнения в build.gradle, например для поддержки корутин и kotlinx.serialization:

dependencies {
  implementation("org.litote.kmongo:kmongo-serialization:4.8.0")
  implementation("org.litote.kmongo:kmongo-coroutine-serialization:4.8.0")
}

Для создания подключения к MongoDB нужно обратиться к функции createClient клиента KMongo. При установке поддержки неблокирующего исполнения от результата можно обратиться к get-методу coroutine для получения асинхронного варианта подключения, например так:

    val client = KMongo.createClient(connectionString = "mongodb://mongoip:27017").coroutine

Полученный объект CoroutineClient позволяет просмотреть список доступных баз данных (listDatabases), подключиться к базе данных getDatabase(name), а также подписаться на поток изменений (watch). Например, для подключения к базе данных test можно использовать следующий вызов:

val database = client.getDatabase("test")

Если клиент был получен как CoroutineClient, то и база данных также будет доступна через обертку CoroutineDatabase. Далее для подключенной базы данных можно выполнять запросы и здесь мы впервые встретимся с необходимостью работы со структурированными объектами. Структура может определяться как любой data-класс и используется при выполнении запросов, а также для определения условий выбора. Например, для выбора всех сообщений со значением числового поля type==1 можно выполнить следующий фрагмент кода:

import org.litote.kmongo.*
import kotlinx.serialization.*

@Serializable
data class TestItem(val type:Int, val message:String)

suspend fun getItems(database:CoroutineDatabase):List<TestItem> {
  val collection = database.getCollection<TestItem>("items")
  return collection.find(TestItem::type eq 1)
}

Аналогично могут быть добавлены новые записи в коллекцию:

collection.insertOne(TestItem(1, "Test"))

Но сложности начинаются при необходимости добавления или извлечения данных со сложными типами данных или произвольной структуры. Для первого сценария есть возможность создавать кодеки, которые смогут интерпретировать произвольные (в том числе, двоичные) типы данных в Mongo в объекты Kotlin. Даты поддерживаются изначально, для остальных можно зарегистрировать кодек через ObjectMappingConfiguration.addCustomCodec для соответствующего типа результатами. А вот со вторым случаем возникает существенная сложность из-за отсутствия механизма сериализации для динамических структур. Предположим, что в коллекции items будет также храниться дополнительное поле meta, в котором будет размещаться map со строковыми ключами и произвольными типами в значении. Казалось бы решением могло бы быть следующее определение:

data class TestItem(val type:Int, val message:String, val meta:Map<String,An

Но сложность тут будет в том, что динамические типы данных не поддерживаются при декодировании ответа Mongo. Как же действовать в такой ситуации? Здесь мы можем использовать объект класса BsonDocument, который позволяет работать с отдельными именованными полями в Mongo-объекте, в том числе, получать список всех существующих полей в документе и определять их тип. BsonDocument может использоваться как при выполнении запроса, так и внутри сериализуемого объекта. Попробуем извлечь структуру данных с неструктурированным полем:

suspend fun getData(db:CoroutineDatabaes):List<TestItem> {
  val items = mutableListOf<TestItem>()
  val collection = db.getCollection<BsonDocument>("items")
  collection.consumeEach { item ->
    val metaData = mutableMapOf<String,Any>()
    val type = item.getInt32("type").value
    val message = item.getString("message").value
    val meta = item["meta"]?.asDocument()
    meta?.keys?.forEach { key ->
      metaData[key] = meta.getValue(key).value
    }
    items.add(TestItem(type, message, metaData))
  }
  return items
}

Здесь мы используем возможности итерации по списку ключей BsonDocument и общий интерфейс для получения значений для любого поля (также может быть возвращен BsonDocument и, в этом случае, можно обработать его рекурсивно). Аналогично может быть сохранен объект с произвольной структурой:

suspend fun addData(db:CoroutineDatabase, data:TestItem) {
  val meta = BsonDocument()
  data.meta.keys.forEach { key ->
    val value = data.meta[key]
    if (value is String) {
      meta[key] = BsonString(value))
    } else if (value is Boolean) {
      meta[key] = BsonBoolean(value))
    } else if (value is Double) {
      meta[it] = BsonDouble(value)
    } else if (value is Int) {
      meta[it] = BsonInt64(value)
    }
  }
  val doc = BsonDocument(
    listOf(
      BsonElement("type", BsonInt32(data.type)),
      BsonElement("message", BsonString(data.message)),
      BsonElement("meta", meta),
    )
  )
  db.getCollection<BsonDocument>("items").insertOne(doc)
}

Таким образом, можно сохранять и извлекать документы произвольной структуры из MongoDB в приложениях на Java/Kotlin. Конечно же, есть возможность создания собственных сериализаторов для kotlinx.serialization / Jackson и кодеков для драйвера, но в большинстве случаев прямого доступа к BsonDocument оказывается достаточным для работы со сложными документами.

---

Завтра в OTUS состоится открытое занятие, которое будет посвящено установке mongo и компаса, также спикер покажет основные операции. Если кому-то интересно — записывайтесь по ссылке.