Identifier

coola.identifier ¶

Provide identifiers: content-derived, time-ordered and human- friendly.

The stable identifiers (generate_stable_content_id, generate_stable_uuid5) are derived from nested data through coola.hashing. The others are generated: time-ordered (ULID, UUIDv7, Snowflake, ObjectId) or checksummed.

Every generator returns a str identifier except SnowflakeIdGenerator.generate, which returns a plain int (a Snowflake ID is defined as a 64-bit integer, e.g. for use as a database BIGINT primary key).

coola.identifier.SnowflakeIdGenerator ¶

Generate Snowflake-style 64-bit identifiers.

An ID packs a 41-bit millisecond timestamp (counted from 2024-01-01), the 10-bit worker_id of the generator and a 12-bit sequence number into an integer of at most 63 bits. IDs minted with the current time are strictly increasing within a generator, from any number of threads. The generator never raises or sleeps because of the clock: when more than 4,096 IDs are requested in the same millisecond, or when the clock moves backward, the timestamp is kept (or advanced by one millisecond) and the sequence goes on, so the timestamp of the IDs may run ahead of the clock until the clock catches up.

Create one generator per process (or worker) and share it: the state is per generator, and there is no module-level generator because the worker_id can only be chosen by the caller.

Warning

The uniqueness of the IDs across processes comes from the worker_id, not from random values. Two generators that use the same worker_id can mint the same ID in the same millisecond. Assigning the worker numbers is the job of the caller (a database row, an environment variable, ...).

Note

A generator cannot be used in the child process of a fork (os.fork, or multiprocessing with the fork start method): the child would continue the sequence of its parent with the same worker_id. generate raises a RuntimeError in the child, which has to create a generator with its own worker_id.

Parameters:

Name	Type	Description	Default
`worker_id`	`int`	The worker ID, in `[0, 1023]`, written in every ID.	required

Raises:

Type	Description
`TypeError`	If `worker_id` is not an `int` (a `bool` is rejected).
`ValueError`	If `worker_id` does not fit in 10 bits.

Example

>>> from coola.identifier import SnowflakeIdGenerator
>>> generator = SnowflakeIdGenerator(worker_id=3)
>>> snowflake_id = generator.generate()
>>> isinstance(snowflake_id, int)
True

coola.identifier.SnowflakeIdGenerator.generate ¶

generate(*, timestamp_ms: int | None = None) -> int

Generate a Snowflake-style 64-bit identifier.

Parameters:

Name	Type	Description	Default
`timestamp_ms`	`int \| None`	The Unix timestamp in milliseconds to encode. If `None` (default), the current time is used and the IDs are strictly increasing. An explicit timestamp is for tests and backfills: it does not touch the state of the generator and the sequence part is random, so such IDs are neither ordered nor guaranteed to be unique.	`None`

Returns:

Type	Description
`int`	A non-negative integer of at most 63 bits.

Raises:

Type	Description
`RuntimeError`	If the generator is used in the child process of a fork.
`TypeError`	If `timestamp_ms` is not an `int` (a `bool` is rejected).
`ValueError`	If `timestamp_ms` is before 2024-01-01 or after the last millisecond of the 41-bit field.

Example

>>> from coola.identifier import SnowflakeIdGenerator
>>> generator = SnowflakeIdGenerator(worker_id=3)
>>> snowflake_id = generator.generate()
>>> isinstance(snowflake_id, int)
True

coola.identifier.extract_object_id_timestamp_s ¶

extract_object_id_timestamp_s(object_id: str) -> int

Extract the Unix timestamp encoded in an ObjectId-style identifier.

Inverse of the encoding done by generate_object_id: decodes the leading 4 bytes of the 24-character hex string back to the Unix timestamp (in seconds) it was created from.

Parameters:

Name	Type	Description	Default
`object_id`	`str`	The identifier string previously returned by `generate_object_id`.	required

Returns:

Type	Description
`int`	The Unix timestamp, in seconds, that was encoded in
`int`	`object_id`.

Raises:

Type	Description
`TypeError`	If `object_id` is not a `str`.
`ValueError`	If `object_id` is not a 24-character lowercase hex string (no whitespace, no upper case, no `bytes`).

Example

>>> from coola.identifier import extract_object_id_timestamp_s, generate_object_id
>>> extract_object_id_timestamp_s(generate_object_id(timestamp_s=1704067200))
1704067200

coola.identifier.extract_snowflake_timestamp_ms ¶

extract_snowflake_timestamp_ms(snowflake_id: int) -> int

Extract the millisecond timestamp encoded in a Snowflake-style identifier.

Inverse of the encoding done by SnowflakeIdGenerator.generate : shifts out the worker ID and sequence fields and adds back the fixed epoch that was subtracted when the ID was minted.

Parameters:

Name	Type	Description	Default
`snowflake_id`	`int`	The identifier previously returned by `SnowflakeIdGenerator.generate`.	required

Returns:

Type	Description
`int`	The Unix timestamp, in milliseconds, that was encoded in
`int`	`snowflake_id`.

Raises:

Type	Description
`TypeError`	If `snowflake_id` is not an `int` (a `bool` is rejected).
`ValueError`	If `snowflake_id` is negative or does not fit in 63 bits.

Example

>>> from coola.identifier import SnowflakeIdGenerator, extract_snowflake_timestamp_ms
>>> snowflake_id = SnowflakeIdGenerator(worker_id=3).generate(timestamp_ms=1704067200000)
>>> extract_snowflake_timestamp_ms(snowflake_id)
1704067200000

coola.identifier.extract_ulid_timestamp_ms ¶

extract_ulid_timestamp_ms(ulid: str) -> int

Extract the millisecond timestamp encoded in a ULID.

Inverse of the encoding done by generate_ulid: decodes the 26-character Crockford Base32 string back to its 128-bit value and returns the top 48 bits, which is the timestamp generate_ulid packed in.

Parameters:

Name	Type	Description	Default
`ulid`	`str`	The ULID string previously returned by `generate_ulid`.	required

Returns:

Type	Description
`int`	The Unix timestamp in milliseconds that was encoded in
`int`	`ulid`.

Raises:

Type	Description
`TypeError`	If `ulid` is not a `str`.
`ValueError`	If `ulid` is not 26 characters long, or contains a character outside the Crockford Base32 alphabet (the case is ignored, `O` is read as `0` and `I`/`L` as `1`), or encodes a value larger than 128 bits.

Example

>>> from coola.identifier import extract_ulid_timestamp_ms, generate_ulid
>>> extract_ulid_timestamp_ms(generate_ulid(timestamp_ms=1704067200000))
1704067200000

coola.identifier.extract_uuid7_timestamp_ms ¶

extract_uuid7_timestamp_ms(uuid7: str) -> int

Extract the millisecond timestamp encoded in a UUIDv7.

Inverse of the encoding done by generate_uuid7: the timestamp is the top 48 bits of the UUID, unaffected by the version/variant bits stamped into the lower 80 bits.

Parameters:

Name	Type	Description	Default
`uuid7`	`str`	The UUIDv7 string previously returned by `generate_uuid7`.	required

Returns:

Type	Description
`int`	The Unix timestamp in milliseconds that was encoded in
`int`	`uuid7`.

Raises:

Type	Description
`TypeError`	If `uuid7` is not a `str` (pass `str(uuid)` for a `uuid.UUID`).
`ValueError`	If `uuid7` is not a UUID in its canonical 36-character form (no braces, no `urn:uuid:` prefix, no missing hyphens), or is not a version 7 UUID of the RFC 9562 variant. The case is ignored.

Example

>>> from coola.identifier import extract_uuid7_timestamp_ms, generate_uuid7
>>> extract_uuid7_timestamp_ms(generate_uuid7(timestamp_ms=1704067200000))
1704067200000

coola.identifier.generate_checksummed_id ¶

generate_checksummed_id(
    *,
    length: int = _DEFAULT_LENGTH,
    group_size: int = _DEFAULT_GROUP_SIZE,
    sep: str = "-",
) -> str

Generate a random identifier with a trailing check symbol.

Draws length random Crockford Base32 characters, appends one check symbol computed from them (a mod-37 checksum, per the Crockford Base32 spec), and groups the result into chunks of group_size characters separated by sep for readability.

The check symbol is always a character of the 32-character alphabet (0-9A-Z without I, L, O and U): when the checksum of a payload is one of the 5 extra symbols of the specification (*~$=U), the payload is drawn again (5 payloads in 37, so 1.16 draws on average). The identifier is therefore safe in a shell, a URL and a spreadsheet, at the cost of 0.21 bit of entropy.

Parameters:

Name	Type	Description	Default
`length`	`int`	The number of random (non-check) characters to generate. Must be positive.	`_DEFAULT_LENGTH`
`group_size`	`int`	The number of characters per group in the formatted output. Must be positive. Pass a number greater than or equal to `length + 1` (or `sep=""`) to disable grouping.	`_DEFAULT_GROUP_SIZE`
`sep`	`str`	The separator inserted between groups. Must be empty or a single character that is not in the Crockford Base32 alphabet (the case is ignored), or `verify_checksummed_id` would not be able to tell a separator character apart from a payload/check one.	`'-'`

Returns:

Type	Description
`str`	A string of `length` random characters plus one check
`str`	symbol, grouped by `group_size` and joined by `sep`.

Raises:

Type	Description
`TypeError`	If `length` or `group_size` is not an `int` (a `bool` is rejected), or `sep` is not a `str`.
`ValueError`	If `length` or `group_size` is not positive, if `length` exceeds 1024, or if `sep` is more than one character long or a character of the Crockford Base32 alphabet.

Example

>>> from coola.identifier import generate_checksummed_id, verify_checksummed_id
>>> checksummed_id = generate_checksummed_id()
>>> verify_checksummed_id(checksummed_id)
True
>>> len(generate_checksummed_id(length=8, group_size=3, sep=" "))
11

coola.identifier.generate_object_id ¶

generate_object_id(
    *, timestamp_s: int | None = None
) -> str

Generate a MongoDB ObjectId style 12-byte identifier.

An ObjectId packs a 32-bit timestamp in seconds followed by 64 random bits into a 24-character lowercase hex string. Because the timestamp is the most significant part, ObjectIds generated later sort (as plain strings) after ObjectIds generated earlier.

Note

When the current time is used (timestamp_s=None), ObjectIds are strictly increasing within a process: ObjectIds generated in the same second (or after the clock moved backward) keep the timestamp and increment the random part by one. The state is per process, protected by a lock, and reset in the child process of a fork. Across processes, ObjectIds generated in the same second are not ordered. When timestamp_s is given, the random part is always random and the state is not touched: two ObjectIds with the same timestamp are not ordered.

Parameters:

Name	Type	Description	Default
`timestamp_s`	`int \| None`	The Unix timestamp in seconds to encode. If `None` (default), the current time is used. Exposed mainly for deterministic testing and for backfilling.	`None`

Returns:

Type	Description
`str`	A 24-character lowercase hex string.

Raises:

Type	Description
`TypeError`	If `timestamp_s` is not an `int` (a `bool` is rejected).
`ValueError`	If `timestamp_s` does not fit in 32 bits (i.e. is negative or exceeds `2**32 - 1`).

Example

>>> from coola.identifier import generate_object_id
>>> object_id = generate_object_id()
>>> len(object_id)
24

coola.identifier.generate_stable_content_id ¶

generate_stable_content_id(
    data: object,
    *,
    registry: HasherRegistry | None = None,
    length: int = 64,
    ignore_unhashable: bool = False,
) -> str

Compute a content-addressed identifier for a nested data structure.

Unlike generate_stable_uuid5, the returned identifier is the raw hash_object digest: it is not reshaped into a UUID, so its collision resistance and length are exactly those of the underlying hash. Use it when more than the 122 bits of a UUID are needed.

Parameters:

Name	Type	Description	Default
`data`	`object`	The data to derive an identifier from. Can be a nested structure such as a `list`, `dict`, or `tuple`.	required
`registry`	`HasherRegistry \| None`	The registry used to resolve hashers for each data type, forwarded to `hash_object`. If `None`, the stable registry (`get_stable_registry`) is used: it has only the built-in hashers, so the identifier does not depend on the installed plugins or on `register_hashers`. Passing another registry opts out of this stability.	`None`
`length`	`int`	The desired length of the returned hex string, forwarded to `hash_object`. Must be an even number between 2 and 128 inclusive. Defaults to `64`.	`64`
`ignore_unhashable`	`bool`	Forwarded to `hash_object`. If `True`, objects for which no hasher is registered are replaced by a deterministic placeholder hash instead of raising an error. Defaults to `False`, which raises a `KeyError` when an unhashable object is encountered.	`False`

Returns:

Type	Description
`str`	A lowercase hex string identifier of the requested `length`.

Raises:

Type	Description
`KeyError`	If `data` (or a nested object within it) has a type for which no hasher is registered and `ignore_unhashable` is `False`.

Example

>>> from coola.identifier import generate_stable_content_id
>>> generate_stable_content_id({"source": "cats.txt", "page": 1})  # doctest: +ELLIPSIS
'...'
>>> generate_stable_content_id(
...     {"page": 1, "source": "cats.txt"}
... ) == generate_stable_content_id({"source": "cats.txt", "page": 1})
True

coola.identifier.generate_stable_uuid5 ¶

generate_stable_uuid5(
    data: object,
    *,
    registry: HasherRegistry | None = None,
    namespace: UUID = _NAMESPACE,
    ignore_unhashable: bool = False,
) -> str

Compute a stable, reproducible UUID for a nested data structure.

Hashes data via generate_stable_content_id (at its maximum, 128-hex-digit length) to guarantee a consistent digest regardless of e.g. mapping insertion order, then derives a deterministic UUID from that digest using uuid.uuid5 under a fixed namespace.

Note

generate_stable_uuid5(a) == generate_stable_uuid5(b) if and only if hash_object(a, length=128) == hash_object(b, length=128) (modulo the astronomically unlikely case of a uuid.uuid5 collision on two different digests). A UUID has 122 random bits, so its collision resistance is bounded by its width (birthday bound 2^61), not by SHA-1 or by the strength of the digest: use generate_stable_content_id when more bits are needed.

Warning

The value returned by generate_stable_uuid5 for a given data is stable only as long as hash_object (and the hashers resolved by registry for the types in data) keep producing the same digest for that data. A change to the hashing algorithms of coola in a future release would silently change the UUIDs produced here. Do not rely on cross-version stability for UUIDs persisted long-term (e.g. as database primary keys) unless you pin coola and pass an explicit, version-controlled registry.

Parameters:

Name	Type	Description	Default
`data`	`object`	The data to derive a UUID from. Can be a nested structure such as a `list`, `dict`, or `tuple`.	required
`registry`	`HasherRegistry \| None`	The registry used to resolve hashers for each data type, forwarded to `hash_object`. If `None`, the stable registry (`get_stable_registry`) is used: it has only the built-in hashers, so the identifier does not depend on the installed plugins or on `register_hashers`. Passing another registry opts out of this stability.	`None`
`namespace`	`UUID`	The UUID namespace passed to `uuid.uuid5`. Defaults to a namespace fixed for this module; pass a different one to derive UUIDs in a separate identifier space (e.g. to avoid collisions with UUIDs minted by another system for unrelated data).	`_NAMESPACE`
`ignore_unhashable`	`bool`	Forwarded to `hash_object`. If `True`, objects for which no hasher is registered are replaced by a deterministic placeholder hash instead of raising an error. Defaults to `False`, which raises a `KeyError` when an unhashable object is encountered.	`False`

Returns:

Type	Description
`str`	A lowercase UUID string of the form
`str`	`'xxxxxxxx-xxxx-5xxx-xxxx-xxxxxxxxxxxx'`.

Raises:

Type	Description
`KeyError`	If `data` (or a nested object within it) has a type for which no hasher is registered and `ignore_unhashable` is `False`.

Example

>>> from coola.identifier import generate_stable_uuid5
>>> generate_stable_uuid5({"source": "cats.txt", "page": 1})  # doctest: +ELLIPSIS
'...'
>>> generate_stable_uuid5({"page": 1, "source": "cats.txt"}) == generate_stable_uuid5(
...     {"source": "cats.txt", "page": 1}
... )
True

coola.identifier.generate_ulid ¶

generate_ulid(*, timestamp_ms: int | None = None) -> str

Generate a ULID (Universally Unique Lexicographically Sortable Identifier).

A ULID packs a 48-bit millisecond timestamp followed by 80 bits of randomness into a 26-character Crockford Base32 string. Because the timestamp is the most significant part, ULIDs generated later sort (lexicographically, as plain strings) after ULIDs generated earlier, unlike uuid.uuid4 which sorts randomly.

Note

When the current time is used (timestamp_ms=None), ULIDs are monotonic within a process, as recommended by the ULID specification: ULIDs generated in the same millisecond (or after the clock moved backward) keep the timestamp and increment the random part by one, so they still sort in the order of generation. The state is per process, protected by a lock, and reset in the child process of a fork. Across processes, ULIDs generated in the same millisecond are not ordered. When timestamp_ms is given, the random part is always random: two ULIDs with the same timestamp are not ordered.

Parameters:

Name	Type	Description	Default
`timestamp_ms`	`int \| None`	The Unix timestamp in milliseconds to encode. If `None` (default), the current time is used and the ULIDs are monotonic (see below). Exposed mainly for deterministic testing and for backfilling.	`None`

Returns:

Type	Description
`str`	A 26-character uppercase Crockford Base32 ULID string.

Raises:

Type	Description
`TypeError`	If `timestamp_ms` is not an `int` (a `bool` is rejected).
`ValueError`	If `timestamp_ms` does not fit in 48 bits (i.e. is negative or exceeds `2**48 - 1`).

Example

>>> from coola.identifier import generate_ulid
>>> ulid = generate_ulid()
>>> len(ulid)
26

coola.identifier.generate_uuid7 ¶

generate_uuid7(*, timestamp_ms: int | None = None) -> str

Generate a UUIDv7 (RFC 9562) identifier.

A UUIDv7 packs a 48-bit millisecond timestamp, followed by the 4-bit version, 12 bits of randomness, the 2-bit variant, and 62 more bits of randomness, into a standard 128-bit UUID layout. Because the timestamp is the most significant part, UUIDv7 values generated later sort (lexicographically, as canonical UUID strings) after UUIDv7 values generated earlier, unlike uuid.uuid4 which sorts randomly.

Note

When the current time is used (timestamp_ms=None), UUIDv7 values are monotonic within a process (the "method 1" counter of RFC 9562): values generated in the same millisecond (or after the clock moved backward) keep the timestamp and increment the 74 random bits by one, so they still sort in the order of generation, which keeps database indexes append-only. The state is per process, protected by a lock, and reset in the child process of a fork. Across processes, values generated in the same millisecond are not ordered. When timestamp_ms is given, the random bits are always random: two values with the same timestamp are not ordered.

Parameters:

Name	Type	Description	Default
`timestamp_ms`	`int \| None`	The Unix timestamp in milliseconds to encode. If `None` (default), the current time is used and the UUIDs are monotonic (see below). Exposed mainly for deterministic testing and for backfilling.	`None`

Returns:

Type	Description
`str`	A lowercase UUID string of the form
`str`	`'xxxxxxxx-xxxx-7xxx-yxxx-xxxxxxxxxxxx'`.

Raises:

Type	Description
`TypeError`	If `timestamp_ms` is not an `int` (a `bool` is rejected).
`ValueError`	If `timestamp_ms` does not fit in 48 bits (i.e. is negative or exceeds `2**48 - 1`).

Example

>>> from coola.identifier import generate_uuid7
>>> uuid7 = generate_uuid7()
>>> len(uuid7)
36

coola.identifier.verify_checksummed_id ¶

verify_checksummed_id(
    identifier: str, *, sep: str = "-"
) -> bool

Verify the check symbol of an identifier from generate_checksummed_id.

Per the Crockford Base32 spec, decoding is case-insensitive and normalizes the characters most often confused when an identifier is hand-transcribed: 'O' with '0', and 'I'/'L' with '1'. identifier is normalized this way before its check symbol is verified, so e.g. a lowercase retype or an 'O' typed for a '0' still verifies correctly. The 5 extra check symbols of the specification (*~$=U) are accepted, so an identifier from another implementation verifies.

Parameters:

Name	Type	Description	Default
`identifier`	`str`	The identifier to verify, in the grouped form returned by `generate_checksummed_id`.	required
`sep`	`str`	The group separator used when `identifier` was generated.	`'-'`

Returns:

Type	Description
`bool`	`True` if the trailing character is a valid check symbol for
`bool`	the characters preceding it, `False` otherwise (including
`bool`	when `identifier` is not a `str`, contains a non-ASCII
`bool`	character or a character outside the extended Crockford Base32
`bool`	alphabet, or is too short to contain a payload and a check
`bool`	symbol).

Raises:

Type	Description
`TypeError`	If `sep` is not a `str`.
`ValueError`	If `sep` is more than one character long or a character of the Crockford Base32 alphabet.

Example

>>> from coola.identifier import generate_checksummed_id, verify_checksummed_id
>>> verify_checksummed_id(generate_checksummed_id())
True
>>> verify_checksummed_id("not-a-valid-id")
False
>>> verify_checksummed_id(generate_checksummed_id().lower())
True