cryomongo

A MongoDB driver written in pure Crystal.

A MongoDB driver in Crystal. No mongo-c-driver. Linux only, tested on MongoDB 8.0 with Crystal 1.21. macOS and Windows are not supported.

zstd needs libzstd. Field encryption needs libmongocrypt 1.20.4. Auto-encryption also needs crypt_shared (mongo_crypt_v1.so). Setup is in Encryption.

For an object-document mapper, see moongoon.

This shard

elbywan/cryomongo is the original driver. Its author is no longer maintaining that shard. This repository, alumna/cryomongo, began as a temporary fork and is now the shard to install.

Current version supports Linux, Crystal 1.21+, and MongoDB 8.0 (wire 25 through 29). It includes the core driver and local-KMS client-side encryption. The public API can still change before 1.0.0.

Not in this version: macOS, Windows, Atlas Search, cloud KMS, MONGODB-AWS, MONGODB-OIDC, and Queryable Encryption prefix / suffix / substring (MongoDB 8.2+). Open work: ROADMAP.md.

Features

CRUD - helpers, collection bulk, client bulkWrite (MongoDB 8.0), let, legacy count, mapReduce, Database#drop / Collection#drop
Aggregation
Read / write concerns and read preference
Authentication - SCRAM-SHA-1/256 (SASLprep on SHA-256 passwords), X509, PLAIN
TLS - tlsCertificateKeyFilePassword; stapled OCSP unless tlsDisableCertificateRevocationCheck
Indexes
GridFS - session:, delete_by_name / rename_by_name, drop
Client-side encryption - local KMS; explicit create_data_key / encrypt / decrypt; auto-encryption with schemaMap and Queryable Encryption encryptedFieldsMap (needs vendored libmongocrypt 1.20.4 and crypt_shared)
Change streams - #next waits; #try_next polls; resume after a labeled getMore error
Tailable cursors and collation
Standalone, replica set, sharded, load-balanced
CSOT - URI / client / database / collection timeoutMS; remaining time becomes maxTimeMS
Wire compression - zlib, snappy, zstd (compressors=)
Command monitoring - sensitive commands are redacted; CMAP subscribe; spec logs MONGODB_LOG_*
Retryable reads and writes
Causal consistency and transactions - pinned mongos; handshake retry on that pin
Versioned API
Raw commands - command, Database#run_command / #run_cursor_command

Generated API pages live in docs/. Regenerate them before trusting that folder.

Installation

Linux. CI tests Crystal 1.21. shard.yml allows >= 1.20.0.

dependencies:
  cryomongo:
    github: alumna/cryomongo

Then shards install.

zstd. Install libzstd-dev (sudo apt-get install libzstd-dev on Debian/Ubuntu). snappy is pure Crystal. zlib comes with Crystal.

The default compile links libmongocrypt and stops if that library is missing. Encryption is how to vendor it, use a system package, or build without it.

Encryption

Field encryption is optional. Pick one row. API samples are in Client-side encryption.

You need	What to install
No field encryption	nothing; compile with `-Dwithout_libmongocrypt`
Explicit encrypt and decrypt	libmongocrypt 1.20.4
Auto-encryption or Queryable Encryption	libmongocrypt 1.20.4 and `mongo_crypt_v1.so`

This driver does not spawn mongocryptd. Cloud KMS is not in 1.0.0-beta.

Without encryption

Add -Dwithout_libmongocrypt to the compile of your program. Mongo::ClientEncryption raises if you construct it. Encrypt specs do not run. The rest of the driver is the same.

Explicit encryption

Link official libmongocrypt 1.20.4, then compile with no extra flag.

From an application, after shards install:

lib/cryomongo/scripts/vendor-libmongocrypt.sh

From this repository:

scripts/vendor-libmongocrypt.sh

The script writes gitignored vendor/libmongocrypt/ next to the driver sources. shards update can remove that directory; run the script again after an update. The Linux archive has no crypto library of its own. The driver registers OpenSSL hooks.

Ubuntu 24.04 libmongocrypt-dev is too old. For a system package >= 1.20.0, set USE_SYSTEM_LIBMONGOCRYPT=true on that compile. pkg-config must find the package. Leave the variable unset when you use the vendored library.

Auto-encryption

Queryable Encryption uses this same pair of libraries. Finish Explicit encryption first, then download crypt_shared.

From an application:

lib/cryomongo/scripts/download-crypt-shared.sh

From this repository:

scripts/download-crypt-shared.sh

The script writes gitignored tmp/mongo_crypt_v1.so and prints CRYPT_SHARED_LIB_PATH=....

Pass that file to every client in the process. Use the .so path, not the directory that contains it. Export CRYPT_SHARED_LIB_PATH, or set extraOptions.cryptSharedLibPath as in the sample. One path for the whole process.

To fetch the MongoDB 8.0 package yourself: ubuntu2204 on Ubuntu 22.04, ubuntu2404 on 24.04 and 26.04 (there is no ubuntu2604 package), x86_64 or aarch64.

Specs skip live auto-encryption when either library is missing.

Usage

Minimal working example

require "cryomongo"

# Create a Mongo client, using a standard mongodb connection string.
client = Mongo::Client.new # defaults to: "mongodb://localhost:27017"

# Get database and collection.
database = client["database_name"]
collection = database["collection_name"]

# Perform crud operations.
result = collection.insert_one({ one: 1 })
# The driver adds `_id` when the document is a BSON. inserted_ids is on the result.
puts result.try(&.inserted_ids)
collection.replace_one({ one: 1 }, { two: 2 })
bson = collection.find_one({ two: 2 })
puts bson.try(&.["two"]) # => 2
collection.delete_one({ two: 2 })
puts collection.count_documents # => 0_i64
client.close

Complex example with serialization

require "cryomongo"

# We take advantage of the BSON serialization capabilities provided by the `bson.cr` shard.
record User,
  name : String,
  banned : Bool? = false,
  _id : BSON::ObjectId = BSON::ObjectId.new,
  creation_date : Time = Time.utc do
  include BSON::Serializable
  include JSON::Serializable
end

# Initialize Client, Database and Collection.
client = Mongo::Client.new
database = client["database"]
users = database["users"]

# We set majority read and write at the Database level.
database.read_concern = Mongo::ReadConcern.new(level: "majority")
database.write_concern = Mongo::WriteConcern.new(w: "majority")

# Drop and recreate the Collection to ensure that we read later only the documents we inserted in this example.
{ Mongo::Commands::Drop, Mongo::Commands::Create }.each do |command|
  database.command(command, name: "users")
rescue e : Mongo::Error::Command
  # ignore the server error, drop will fail if the collection has not been created before.
end

# Insert User structures that are automatically serialized to BSON.
users.insert_many(["John", "Jane"].map { |name| User.new(name: name) })

# Fetch a Cursor pointing to the users collection.
# `#each` (and `to_a`) close the cursor when iteration ends.
# You can also pass a block to `find`. Call `#close` if you use `#next` yourself.
cursor = users.find

# Iterate the cursor and use `.of(User)` to deserialize as the cursor gets iterated.
# Then push the users into an array that gets pretty printed.
puts cursor.of(User).to_a.to_pretty_json
# => [
#   {
#     "name": "John",
#     "banned": false,
#     "_id": {
#       "$oid": "f2001c5fb0a33e0264e2ea05"
#     },
#     "creation_date": "2020-07-25T09:52:50Z"
#   },
#   {
#     "name": "Jane",
#     "banned": false,
#     "_id": {
#       "$oid": "f2001c5fb0a33e0264e2ea07"
#     },
#     "creation_date": "2020-07-25T09:52:50Z"
#   }
# ]

Conventions

Methods and arguments names are in snake case.
Object arguments can usually be passed as a NamedTuple, Hash, BSON::Serializable or a BSON instance.

Connection

require "cryomongo"

# Mongo::Client monitors the cluster, routes commands, and owns the pools.

# A client can be instantiated using a standard mongodb connection string.
# Replica set:
#   Mongo::Client.new("mongodb://localhost:27017/?replicaSet=rs0")
# Load balancer:
#   Mongo::Client.new("mongodb://localhost:8000/?loadBalanced=true")
# Stable API (not a URI option):
#   options = Mongo::Options.new
#   options.server_api = Mongo::ServerApi.new(version: "1", strict: true)
#   Mongo::Client.new("mongodb://localhost:27017", options)
# CSOT. One deadline for selection, checkout, and maxTimeMS. 0 means no timeout.
#   Mongo::Client.new("mongodb://localhost:27017/?timeoutMS=5000")
#   collection.find({one: 1}, timeout_ms: 500)

# Client options can be passed as query parameters…
client = Mongo::Client.new("mongodb://address:port/database?appname=MyApp")
# …or with a Mongo::Options instance…
options = Mongo::Options.new(appname: "MyApp")
client = Mongo::Client.new("mongodb://address:port/database", options)
# …or both.

# Instantiate objects to interact with a specific database or a collection…
database   = client["database_name"]
collection = database["collection_name"]

# …or using `default_database` if the connection uri string contains a default auth database component ("/database").
if database = client.default_database
  collection = database["collection_name"]
end

# One client per process is the usual shape. Call #close when you are done with it.
client.close

# To enable SSL/TLS, use the `tls` option, alongside `tlsCAFile` and `tlsCertificateKeyFile`.
# Encrypted PEM keys use `tlsCertificateKeyFilePassword`.
uri = "mongodb://localhost:27017/?tls=true&tlsCAFile=./ca.crt&tlsCertificateKeyFile=./client.pem&tlsCertificateKeyFilePassword=secret"
ssl_client = Mongo::Client.new uri

# Wire compression. The driver uses the first name that the server also has.
uri = "mongodb://localhost:27017/?compressors=snappy,zlib,zstd"
client = Mongo::Client.new uri

Links

Authentication

Supported: SCRAM-SHA-1, SCRAM-SHA-256 (SASLprep on the password), X509, and PLAIN.

Not supported yet: MONGODB-AWS, MONGODB-OIDC.

require "cryomongo"

# Username and password. The server picks SCRAM-SHA-1 or SCRAM-SHA-256.
client = Mongo::Client.new("mongodb://username:password@localhost:27017")

# Or set the mechanism:
# mongodb://username:password@localhost:27017/?authMechanism=SCRAM-SHA-256
# mongodb://localhost:27017/?authMechanism=MONGODB-X509&tls=true
# mongodb://user:pass@localhost:27017/?authMechanism=PLAIN

Basic operations

require "cryomongo"

client = Mongo::Client.new

# Most CRUD operations are performed at collection-level.
collection = client["database_name"]["collection_name"]

# The examples below are very basic, but the methods can accept all the options documented in the MongoDB manual.

## Create

# Insert a single document
collection.insert_one({ key: "value" })
# Insert multiple documents
collection.insert_many((1..100).map { |i| { count: i } })

# To track the _id, generate and pass it as a property
id = BSON::ObjectId.new
collection.insert_one({ _id: id, key: "value" })

## Read

# Find a single document
document = collection.find_one({ _id: id })
document.try { |d| puts d.to_json }

# Find multiple documents.
cursor = collection.find({ qty: { "$gt": 4 }})
elements = cursor.to_a # to_a uses #each, which closes the cursor
# Call #close yourself only when you stop early with #next.

# Collation applies to this operation only.
collection.find({name: "jose"}, collation: Mongo::Collation.new(locale: "en", strength: 2)).to_a

# Tailable awaitData. max_time_ms is the getMore wait.
tailable = collection.find({ts: {"$gt": 1}}, tailable: true, await_data: true, max_time_ms: 1000)
tailable.close

## Update

# Replace a single document.
collection.replace_one({ name: "John" }, { name: "Jane" })
# Update a single document.
collection.update_one({ name: "John" }, { "$set": { name: "Jane" }})
# Update multiple documents
collection.update_many({ name: { "$in": ["John", "Jane"] }}, { "$set": { name: "Jules" }})
# Find one document and replace it
document = collection.find_one_and_replace({ name: "John" }, { name: "Jane" })
puts document.try &.["name"]
# Find one document and update it
document = collection.find_one_and_update({ name: "John" }, { "$set": { name: "Jane" }})
puts document.try &.["name"]

## Delete

# Delete one document
collection.delete_one({ age: 20 })
# Delete multiple documents
collection.delete_many({ age: { "$lt": 18 }})
# find_one_and_delete
document = collection.find_one_and_delete({ age: { "$lt": 18 }})
puts document.try &.["age"]

# Drop the collection (NamespaceNotFound is ignored) or the database.
collection.drop
client["database_name"].drop

# Aggregate

# Perform an aggregation pipeline query
cursor = collection.aggregate([
  {"$match": { status: "available" }},
  {"$limit": 5},
])
cursor.try &.each { |bson| puts bson.to_json }

# let variables (also on find, updates, deletes, find_one_and_*, bulk_write)
cursor = collection.aggregate(
  [{"$match": {"$expr": {"$eq": ["$status", "$$st"]}}}],
  let: { st: "available" },
)

# Map-Reduce (not retryable). Crystal uses output: because out is reserved.
inline = collection.map_reduce(
  "function() { emit(this.status, 1) }",
  "function(key, values) { return Array.sum(values) }",
  output: {inline: 1},
)

# Distinct collection values
values = collection.distinct(
  key: "field",
  filter: { age: { "$gt": 18 }}
)

# Documents count (returns Int64)
counter = collection.count_documents({ age: { "$lt": 18 }})

# Estimated count (also Int64)
counter = collection.estimated_document_count

# Legacy count command (prefer count_documents). Empty filter omits query.
counter = collection.count({ age: { "$lt": 18 }})

# Raw command (not retryable; database read/write concern is not applied).
database = client["database_name"]
ping = database.run_command({ping: 1})
puts ping["ok"] # => 1.0

# Command that returns a cursor. getMore stays on the same server.
cursor = database.run_cursor_command({find: "collection_name", batchSize: 2}, batch_size: 2)
cursor.each { |doc| puts doc.to_json }

Links

Bulk operations

require "cryomongo"

client = Mongo::Client.new

# A Bulk object can be initialized by calling `.bulk` on a collection.
collection = client["database_name"]["collection_name"]
bulk = collection.bulk
# A bulk is ordered by default.
bulk.ordered? # => true

500.times do |idx|
  # Build the queries by calling bulk methods multiple times.
  bulk.insert_one({number: idx})
  bulk.delete_many({number: {"$lt": 450}})
  bulk.replace_one({ number: idx }, { number: idx + 1})
end

# Execute all the queries and return an aggregated result.
pp bulk.execute(write_concern: Mongo::WriteConcern.new(w: 1))

Client bulkWrite (MongoDB 8.0) can write to more than one namespace in one command:

client = Mongo::Client.new
result = client.bulk_write([
  Mongo::ClientBulk::InsertOne.new("database_name.collection_name", {number: 1}),
  Mongo::ClientBulk::DeleteOne.new("database_name.other", {number: 1}),
])
puts result.inserted_count

Links

Indexes

require "cryomongo"

client = Mongo::Client.new
collection = client["database_name"]["collection_name"]

# Create one index without options…
collection.create_index(
  keys: {
    "a":  1,
    "b":  -1,
  }
)
# or with options (snake_cased)…
collection.create_index(
  keys: {
    "a":  1,
    "b":  -1,
  },
  options: {
    unique: true
  }
)
# and optionally specify the name.
collection.create_index(
  keys: {
    "a":  1,
    "b":  -1,
  },
  options: {
    name: "index_name",
  }
)

# Several indexes in one command. String keys keep every model the same Crystal type.
collection.create_indexes([
  {keys: {"a" => 1}, options: {unique: true}},
  {keys: {"c" => 1}, options: {unique: false}},
])
collection.create_index(keys: {"b" => 1}, options: {expire_after_seconds: 3600})

collection.list_indexes.try(&.each { |index| puts index["name"] })
collection.drop_index("index_name")

Links

Mongo::Collection

GridFS

require "cryomongo"

client = Mongo::Client.new
database = client["database_name"]

# A GridFS bucket belongs to a database.
gridfs = database.grid_fs

# Upload (using File.open ensures the file descriptor is closed automatically).
# All GridFS methods accept session: if you need a transaction or causal reads.
id = File.open("file.txt") do |file|
  gridfs.upload_from_stream("file.txt", file)
end

# Download
stream = IO::Memory.new
gridfs.download_to_stream(id, stream)
stream.rewind
puts stream.gets_to_end

# Find
files = gridfs.find({
  length: {"$gte": 5000},
})
files.each do |file|
  puts file.filename
end

# Rename, then delete. delete_by_name is the other way to remove a file.
gridfs.rename(id, "new.txt")
gridfs.rename_by_name("new.txt", "newer.txt")
gridfs.delete(id)
# gridfs.delete_by_name("newer.txt")

# Drop the files and chunks collections
gridfs.drop

# And many more methods… (check the link below.)

Links

Mongo::GridFS::Bucket

Client-side encryption

Local KMS only in 1.0.0-beta, including named local:name. The master key is 96 bytes (binary or base64). Ciphertext is BSON binary subtype 0x06. Call #close. The GC must not destroy the libmongocrypt handle. Library setup is in Encryption.

Auto-encryption needs crypt_shared and applies to collection commands. Prefer a local schemaMap. Other JSON Schema rules in that map are not enforced and raise. The user needs listCollections. Auto-encryption lowers the maximum write batch size. Set cryptSharedLibPath or CRYPT_SHARED_LIB_PATH to the .so file, not a directory, and use the same path for every client in the process. Cloud KMS and MongoDB 8.2+ text search stay out.

require "cryomongo"
require "random/secure"

uri = "mongodb://localhost:27017"
client = Mongo::Client.new(uri)
master_key = Random::Secure.random_bytes(96)
kms = BSON.build do |bson|
  bson.document("local") do
    bson["key"] = BSON::Binary.new(BSON::Binary::SubType::Generic, master_key)
  end
end

encryption = Mongo::ClientEncryption.new(
  client,
  key_vault_namespace: "keyvault.datakeys",
  kms_providers: kms
)

key_id = encryption.create_data_key("local")
encrypted = encryption.encrypt(
  "secret",
  algorithm: Mongo::ClientEncryption::ALGORITHM_RANDOM,
  key_id: key_id
)
plain = encryption.decrypt(encrypted) # => "secret"
# Key vault: get_key, get_keys, delete_key, add_key_alt_name,
# remove_key_alt_name, get_key_by_alt_name, rewrap_many_data_key

The deterministic algorithm is Mongo::ClientEncryption::ALGORITHM_DETERMINISTIC. Same plaintext then gives the same ciphertext.

The next two samples continue this program (uri, kms, key_id). Call #close on the encryption handle and the client when the program ends.

Auto-encryption encrypts marked fields on write and decrypts them on read:

schema = BSON.build do |bson|
  bson.document("hr.people") do
    bson["bsonType"] = "object"
    bson.document("properties") do
      bson.document("ssn") do
        bson.document("encrypt") do
          bson["bsonType"] = "string"
          bson["algorithm"] = Mongo::ClientEncryption::ALGORITHM_DETERMINISTIC
          bson.array("keyId") { bson["0"] = key_id }
        end
      end
    end
  end
end

extra = BSON.build do |bson|
  bson["cryptSharedLibPath"] = "/usr/local/lib/mongo_crypt_v1.so"
  bson["cryptSharedLibRequired"] = true
end

auto = Mongo::Client.new(
  uri,
  auto_encryption: Mongo::AutoEncryption.new(
    key_vault_namespace: "keyvault.datakeys",
    kms_providers: kms,
    schema_map: schema,
    extra_options: extra
  )
)
auto["hr"]["people"].insert_one({ssn: "123-45-6789", name: "Ada"})
auto.close

Do not encrypt _id unless that is the goal. Key-vault commands on an auto-encrypting client bypass encryption. bypass_auto_encryption: true still decrypts. bypass_query_analysis: true skips crypt_shared. This driver does not spawn mongocryptd.

Queryable Encryption (MongoDB 8.0 equality and range) uses encrypted_fields_map, not schemaMap. Do not put one collection in both. create_encrypted_collection fills null keyId values, then creates the collection plus ESC and ECOC. Pass the returned encryptedFields to a new client. compact_structured_encryption_data uses that client. A standalone cannot host these collections.

ef = BSON.build do |bson|
  bson.array("fields") do
    bson.document("0") do
      bson["path"] = "ssn"
      bson["bsonType"] = "string"
      bson["keyId"] = nil
      bson.document("queries") { bson["queryType"] = "equality" }
      # Range uses queryType "range" on a number or date, plus contention, trimFactor, and sparsity.
    end
  end
end

_coll, filled = encryption.create_encrypted_collection(
  client["hr"],
  "people",
  encrypted_fields: ef,
  kms_provider: "local"
)
encryption.close
client.close

qe_map = BSON.build { |bson| bson["hr.people"] = filled }
qe = Mongo::Client.new(
  uri,
  auto_encryption: Mongo::AutoEncryption.new(
    key_vault_namespace: "keyvault.datakeys",
    kms_providers: kms,
    encrypted_fields_map: qe_map,
    extra_options: extra
  )
)
qe["hr"]["people"].insert_one({ssn: "123-45-6789", name: "Ada"})
found = qe["hr"]["people"].find_one({ssn: "123-45-6789"})
qe["hr"]["people"].compact_structured_encryption_data
qe.close

Change streams

require "cryomongo"

# Change streams can watch a client, database or collection for change.
# This code snippet will focus on watching a single collection.

client = Mongo::Client.new
collection = client["database_name"]["collection_name"]

spawn do
  cursor = collection.watch(
    [
      {"$match": {"operationType": "insert"}},
    ],
    max_await_time_ms: 10000
  )
  # `#each` and `#next` wait while the stream is open. An empty getMore does not stop.
  # `#try_next` is one poll and updates `resume_token`.
  # `#of(BSON)` yields `ChangeStream::Document(BSON)`.
  begin
    cursor.of(BSON).each do |doc|
      puts doc.document_key
      puts doc.full_document.try(&.to_json)
    end
  ensure
    cursor.close
  end
end

100.times do |i|
  collection.insert_one({count: i})
end

sleep

Links

Raw commands

require "cryomongo"

# Commands can be run on a client, database or collection depending on the command target.

client = Mongo::Client.new

# Call the `.command` method to run a command against the server.
# The first argument is a `Mongo::Commands` sub-class, followed by the mandatory arguments
# and finally an *options* named tuple containing the optional parameters in snake_case.
result = client.command(Mongo::Commands::ServerStatus, options: {
  repl: 0
})
puts result.to_json

# The .command method can also be called against a Database…
client["database"].command(Mongo::Commands::Create, name: "collection")
client["database"].command(Mongo::Commands::Drop, name: "collection")
# …or a Collection.
client["database"]["collection"].command(Mongo::Commands::Validate)

Links

Concerns and Preference

require "cryomongo"

# Instantiate Read/Write Concerns and Preference
read_concern = Mongo::ReadConcern.new(level: "majority")
write_concern = Mongo::WriteConcern.new(w: 1, j: true)
read_preference = Mongo::ReadPreference.new(mode: "primary")

# They can be set on the client, the database, or the collection…
client = Mongo::Client.new
database = client["database_name"]
collection = database["collection_name"]

client.read_concern = read_concern
database.write_concern = write_concern
collection.read_preference = read_preference

# …or by passing an extra argument when calling a method.
collection.find(
  filter: { key: "value" },
  read_concern:  Mongo::ReadConcern.new(level: "local"),
  read_preference: Mongo::ReadPreference.new(mode: "secondary")
)

Links

Commands and SDAM Monitoring

require "cryomongo"

client = Mongo::Client.new

# 1. Command Monitoring Subscriber
# Tracks the execution of database commands (e.g., find, insert, aggregate).
# Sensitive commands (authenticate, saslStart, createUser, …) are redacted.

cmd_subscription = client.subscribe_commands { |event|
  case event
  when Mongo::Monitoring::Commands::CommandStartedEvent
    Log.info { "COMMAND.#{event.command_name} #{event.address} STARTED: #{event.command.to_json}" }
  when Mongo::Monitoring::Commands::CommandSucceededEvent
    Log.info { "COMMAND.#{event.command_name} #{event.address} COMPLETED (#{event.duration}s)" }
  when Mongo::Monitoring::Commands::CommandFailedEvent
    Log.info { "COMMAND.#{event.command_name} #{event.address} FAILED: #{event.failure.inspect} (#{event.duration}s)" }
  end
}

# 2. SDAM (Server Discovery and Monitoring) Subscriber
# Tracks the lifecycle and topology changes of the MongoDB cluster.

sdam_subscription = client.subscribe_sdam { |event|
  case event
  when Mongo::Monitoring::SDAM::ServerDescriptionChangedEvent
    Log.info { "SERVER.#{event.address} CHANGED: #{event.previous_description.type} -> #{event.new_description.type}" }
  when Mongo::Monitoring::SDAM::TopologyDescriptionChangedEvent
    Log.info { "TOPOLOGY CHANGED: #{event.previous_description.type} -> #{event.new_description.type}" }
  when Mongo::Monitoring::SDAM::ServerHeartbeatStartedEvent
    Log.info { "HEARTBEAT.#{event.address} STARTED awaited=#{event.awaited}" }
  when Mongo::Monitoring::SDAM::ServerHeartbeatSucceededEvent
    Log.info { "HEARTBEAT.#{event.address} OK awaited=#{event.awaited} (#{event.duration})" }
  when Mongo::Monitoring::SDAM::ServerHeartbeatFailedEvent
    Log.info { "HEARTBEAT.#{event.address} FAILED awaited=#{event.awaited}: #{event.failure}" }
  when Mongo::Monitoring::SDAM::ServerClosedEvent
    Log.info { "SERVER.#{event.address} REMOVED FROM TOPOLOGY" }
  end
}

# 3. Connection pool (CMAP) subscriber
# Tracks pool create / ready / clear / close and checkout.

cmap_subscription = client.subscribe_cmap { |event|
  case event
  when Mongo::Monitoring::CMAP::PoolClearedEvent
    Log.info { "POOL.#{event.address} CLEARED interrupt=#{event.interrupt_in_use_connections}" }
  when Mongo::Monitoring::CMAP::ConnectionCheckedOutEvent
    Log.info { "POOL.#{event.address} CHECKED OUT id=#{event.connection_id}" }
  when Mongo::Monitoring::CMAP::ConnectionClosedEvent
    Log.info { "POOL.#{event.address} CLOSED id=#{event.connection_id} reason=#{event.reason}" }
  end
}

# Make some queries…
client["database_name"]["collection_name"].find({ hello: "world" })

# …and eventually at some point, unsubscribe the loggers.
client.unsubscribe_commands(cmd_subscription)
client.unsubscribe_sdam(sdam_subscription)
client.unsubscribe_cmap(cmap_subscription)

Optional spec logs (off unless you set env). MONGODB_LOG_ALL=debug turns every component on. Per component: MONGODB_LOG_COMMAND, MONGODB_LOG_TOPOLOGY, MONGODB_LOG_CONNECTION (debug is the usual value). MONGODB_LOG_PATH is stdout, stderr, or a file (default stderr). MONGODB_LOG_MAX_DOCUMENT_LENGTH truncates command and reply JSON (default 1000). MONGODB_LOG_SERVER_SELECTION is accepted; this driver does not emit those messages yet.

# Example: MONGODB_LOG_COMMAND=debug MONGODB_LOG_PATH=stderr crystal run app.cr

Links

Causal Consistency

require "cryomongo"

client = Mongo::Client.new
# It is important to ensure that both read and writes are performed with "majority" concern.
# See: https://docs.mongodb.com/manual/core/causal-consistency-read-write-concerns/
client.read_concern = Mongo::ReadConcern.new(level: "majority")
client.write_concern = Mongo::WriteConcern.new(w: "majority")

# Reusing the original Mongodb example.
# See: https://docs.mongodb.com/manual/core/read-isolation-consistency-recency/#examples

current_date = Time.utc
items_collection = client["test"]["items"]

# MongoDB enables causal consistency in client sessions by default.
# This is the block syntax that creates, ends and passes the session to collection methods automatically.
items_collection.with_session do |items|
  # Using a causally consistent session ensures that the update occurs before the insert.
  items.update_one(
    { sku: "111", end: { "$exists": false } },
    { "$set": { end: current_date }}
  )
  items.insert_one({ sku: "nuts-111", name: "Pecans", start: current_date })
  puts items.find.to_a.to_pretty_json
end

client.close

Links

Transactions

require "cryomongo"

# Initialize Client and Database instances.
client = Mongo::Client.new
database = client["db"]
collection = database["collection"]

# Create the collection.
{Mongo::Commands::Drop, Mongo::Commands::Create}.each do |command|
  database.command(command, name: "collection")
rescue e : Mongo::Error::Command
  # ignore the server error, drop will fail if the collection has not been created before.
end

# Set read and write concerns to perform isolated transactions.
# See: https://docs.mongodb.com/master/core/transactions/#transactions-and-sessions
transaction_options = Mongo::Session::TransactionOptions.new(
  read_concern: Mongo::ReadConcern.new(level: "snapshot"),
  write_concern: Mongo::WriteConcern.new(w: "majority")
)

# There are two ways to perform transactions:

collection.with_session(default_transaction_options: transaction_options) do |collection, session|
  puts collection.find.to_a.to_json # => "[]"

  # 1. by calling the `with_transaction` method.

  # `with_transaction` will commit after the block ends.
  # if the block raises, the transaction will be aborted.
  session.with_transaction do
    collection.insert_one({_id: 1})
    collection.insert_one({_id: 2})
  end
  puts collection.find.to_a.to_json # => [{"_id":1},{"_id":2}]

  # The transaction below will be aborted because the block raises an Exception.
  begin
    session.with_transaction do
      collection.insert_one({_id: 3})
      raise "Interrupted!"
      collection.insert_one({_id: 4})
    end
  rescue e
    puts e # => Interrupted!
  end
  puts collection.find.to_a.to_json # => [{"_id":1},{"_id":2}]

  # 2. by calling the `start_transaction`, `commit_transaction` and `abort_transaction` methods.
  session.start_transaction
  collection.insert_one({_id: 3})
  # The transaction is isolated, reading outside of the session scope does not return documents impacted by the transaction…
  puts database["collection"].find.to_a.to_json # => [{"_id":1},{"_id":2}]
  # but reading within the session scope does.
  puts collection.find.to_a.to_json # => [{"_id":1},{"_id":2},{"_id":3}]
  session.commit_transaction
  # The transaction is now committed and visible outside of the transaction scope.
  puts collection.find.to_a.to_json             # => [{"_id":1},{"_id":2},{"_id":3}]
  puts database["collection"].find.to_a.to_json # => [{"_id":1},{"_id":2},{"_id":3}]
end

Links

Benchmarks

See BENCHMARK.md (how to run, then the numbers). BSON-only: crystal run bench/driver_bench.cr. Live tasks need MONGODB_URI. A number you can quote: shards build --release driver_bench then BENCH_FULL=1. JSON history: bench/results/.

Contributing

Fork it (https://github.com/alumna/cryomongo/fork)
Create your feature branch (git checkout -b my-new-feature)
Commit your changes (git commit -am 'Add some feature')
Push to the branch (git push origin my-new-feature)
Create a new Pull Request

Specs use crystal spec -Dpreview_mt -Dexecution_context and CRYSTAL_WORKERS=2. GitHub runs that on Ubuntu 22.04, 24.04, and 26.04, x64 and arm64, for standalone, replica set, sharded, and load-balanced. A Darwin compile raises.

Contributors

elbywan - original author
paulocoghi - maintainer

Credit

Icon made by Smashicons from www.flaticon.com.

Repository

cryomongo

Owner

alumna

Statistic

1
0
0
1
6
8 days ago
July 13, 2026

License

MIT License

Links

Synced at

Fri, 09 Oct 2026 22:40:09 GMT

Languages

Crystal 95.62% Shell 4.38%

cryomongo v1.0.1

cryomongo

A MongoDB driver written in pure Crystal.

Contents

This shard

Features

Installation

Encryption

Without encryption

Explicit encryption

Auto-encryption

Usage

Minimal working example

Complex example with serialization

Conventions

Connection

Authentication

Basic operations

Bulk operations

Indexes

GridFS

Client-side encryption

Change streams

Raw commands

Concerns and Preference

Commands and SDAM Monitoring

Causal Consistency

Transactions

Benchmarks

Contributing

Contributors

Credit