Unique ID Generation

import uuid

# Generate a UUID v4 — no coordination needed
id = uuid.uuid4()
print(id)   # e.g., "a1b2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d"
print(len(str(id)))  # 36 characters

import time
import threading

class SnowflakeGenerator:
    """Twitter-style 64-bit unique ID generator."""

    # Custom epoch: Jan 1, 2020 00:00:00 UTC
    EPOCH = 1577836800000

    MACHINE_ID_BITS = 10
    SEQUENCE_BITS = 12

    MAX_MACHINE_ID = (1 << MACHINE_ID_BITS) - 1   # 1023
    MAX_SEQUENCE = (1 << SEQUENCE_BITS) - 1         # 4095

    MACHINE_ID_SHIFT = SEQUENCE_BITS                # 12
    TIMESTAMP_SHIFT = SEQUENCE_BITS + MACHINE_ID_BITS  # 22

    def __init__(self, machine_id: int):
        if machine_id < 0 or machine_id > self.MAX_MACHINE_ID:
            raise ValueError(f"Machine ID must be 0–{self.MAX_MACHINE_ID}")
        self.machine_id = machine_id
        self.sequence = 0
        self.last_timestamp = -1
        self.lock = threading.Lock()

    def generate(self) -> int:
        with self.lock:
            timestamp = self._current_time()

            if timestamp == self.last_timestamp:
                # Same millisecond — increment sequence
                self.sequence = (self.sequence + 1) & self.MAX_SEQUENCE
                if self.sequence == 0:
                    # Sequence exhausted — wait for next millisecond
                    timestamp = self._wait_next_ms(timestamp)
            elif timestamp > self.last_timestamp:
                self.sequence = 0
            else:
                # Clock moved backward — refuse to generate
                raise ClockMovedBackwardError(
                    f"Clock moved backward by "
                    f"{self.last_timestamp - timestamp}ms"
                )

            self.last_timestamp = timestamp

            return (
                ((timestamp - self.EPOCH) << self.TIMESTAMP_SHIFT)
                | (self.machine_id << self.MACHINE_ID_SHIFT)
                | self.sequence
            )

    def _current_time(self) -> int:
        return int(time.time() * 1000)

    def _wait_next_ms(self, last_ts: int) -> int:
        ts = self._current_time()
        while ts <= last_ts:
            ts = self._current_time()
        return ts


class ClockMovedBackwardError(Exception):
    pass

def extract_timestamp(snowflake_id: int, epoch=1577836800000) -> float:
    """Extract Unix timestamp from a Snowflake ID."""
    timestamp_ms = (snowflake_id >> 22) + epoch
    return timestamp_ms / 1000

# Example:
gen = SnowflakeGenerator(machine_id=42)
sid = gen.generate()
print(f"ID: {sid}")
print(f"Created at: {extract_timestamp(sid)}")
print(f"Machine: {(sid >> 12) & 0x3FF}")
print(f"Sequence: {sid & 0xFFF}")

import os
import time

# Crockford Base32 encoding table
ENCODING = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"

def generate_ulid() -> str:
    """Generate a ULID: 48-bit timestamp + 80-bit random."""
    # Timestamp: milliseconds since Unix epoch
    timestamp_ms = int(time.time() * 1000)

    # Encode 48-bit timestamp as 10 Base32 characters
    ts_chars = []
    for _ in range(10):
        ts_chars.append(ENCODING[timestamp_ms & 0x1F])
        timestamp_ms >>= 5
    ts_part = "".join(reversed(ts_chars))

    # 80 bits of randomness as 16 Base32 characters
    random_bytes = os.urandom(10)  # 10 bytes = 80 bits
    random_int = int.from_bytes(random_bytes, "big")
    rand_chars = []
    for _ in range(16):
        rand_chars.append(ENCODING[random_int & 0x1F])
        random_int >>= 5
    rand_part = "".join(reversed(rand_chars))

    return ts_part + rand_part

# Example:
ulid = generate_ulid()
print(ulid)       # e.g., "01HZXK5P9RQCV3Z8KNTDMJ5WGE"
print(len(ulid))  # 26 characters

-- PostgreSQL sequence
CREATE SEQUENCE order_id_seq START 1 INCREMENT 1;

-- Use in a table
CREATE TABLE orders (
    id BIGINT DEFAULT nextval('order_id_seq') PRIMARY KEY,
    ...
);

-- Or explicitly
INSERT INTO orders (id, ...) VALUES (nextval('order_id_seq'), ...);

class BlockAllocator:
    """Pre-allocate ID blocks from a central sequence."""

    def __init__(self, db, block_size=1000):
        self.db = db
        self.block_size = block_size
        self.current = 0
        self.ceiling = 0

    async def next_id(self) -> int:
        if self.current >= self.ceiling:
            await self._allocate_block()
        id = self.current
        self.current += 1
        return id

    async def _allocate_block(self):
        """Atomically reserve a block of IDs by bumping a counter row.

        Backed by a one-row-per-namespace table:
            CREATE TABLE id_blocks (
                name        TEXT PRIMARY KEY,
                next_start  BIGINT NOT NULL DEFAULT 1
            );

        The UPDATE acquires a row lock; concurrent allocators serialize
        on it and each gets a distinct, contiguous block.
        """
        row = await self.db.query_one(
            """
            UPDATE id_blocks
               SET next_start = next_start + %s
             WHERE name = %s
            RETURNING next_start - %s AS block_start
            """,
            (self.block_size, "order_ids", self.block_size),
        )
        self.current = row["block_start"]
        self.ceiling = self.current + self.block_size