From 2e9494fc6f02c490a23328469f66d12abfd0cfcf Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Feb 2026 18:34:43 +0000 Subject: [PATCH 1/3] Add lowercase base36 encoding and fix flaky smoke test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New API for lowercase-only output (0-9a-z, 31 characters): - KSUID.to_base36() / KSUID.from_base36() instance/class methods - generate_lowercase() — sortable KSUID as lowercase string - generate_token_lowercase() — secure random token as lowercase string - from_base36() — module-level convenience to parse base36 strings - Full base36 encode/decode with overflow validation Also: - Fix flaky __main__ smoke test: replaced 1ms sleep (insufficient for 1-second timestamp resolution) with deterministic timestamp comparison - Extract _BASE62_STRING_LENGTH / _BASE36_STRING_LENGTH constants 49 tests passing (15 new for lowercase). https://claude.ai/code/session_01PVPVUNWhpxVa3xbDDBnwp2 --- __init__.py | 117 ++++++++++++++++++++++++++++++++++++++++++++++---- test_ksuid.py | 114 +++++++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 217 insertions(+), 14 deletions(-) diff --git a/__init__.py b/__init__.py index d45fd70..56503cf 100644 --- a/__init__.py +++ b/__init__.py @@ -26,7 +26,12 @@ from typing import Union, Optional __version__ = "1.0.0" -__all__ = ["KSUID", "generate", "generate_token", "from_string", "from_bytes"] +__all__ = [ + "KSUID", + "generate", "generate_lowercase", + "generate_token", "generate_token_lowercase", + "from_string", "from_base36", "from_bytes", +] # KSUID epoch (May 13, 2014 16:53:20 UTC) EPOCH = 1400000000 @@ -36,9 +41,15 @@ PAYLOAD_LENGTH = 16 # 16 bytes for random payload TOTAL_LENGTH = TIMESTAMP_LENGTH + PAYLOAD_LENGTH # 20 bytes total -# Base62 alphabet for encoding +# Base62 alphabet for encoding (mixed-case) BASE62_ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" BASE62_BASE = len(BASE62_ALPHABET) +_BASE62_STRING_LENGTH = 27 # 20 bytes in base62 + +# Base36 alphabet for lowercase encoding +BASE36_ALPHABET = "0123456789abcdefghijklmnopqrstuvwxyz" +BASE36_BASE = len(BASE36_ALPHABET) +_BASE36_STRING_LENGTH = 31 # 20 bytes in base36 class KSUID: @@ -92,13 +103,34 @@ def from_string(cls, ksuid_str: str) -> 'KSUID': Returns: KSUID instance """ - if len(ksuid_str) != 27: - raise ValueError("KSUID string must be exactly 27 characters") - + if len(ksuid_str) != _BASE62_STRING_LENGTH: + raise ValueError(f"KSUID string must be exactly {_BASE62_STRING_LENGTH} characters") + # Decode from base62 decoded_bytes = _base62_decode(ksuid_str) return cls.from_bytes(decoded_bytes) + @classmethod + def from_base36(cls, ksuid_str: str) -> 'KSUID': + """ + Create a KSUID from a lowercase base36 string representation. + + Args: + ksuid_str: Base36-encoded KSUID string (31 characters) + + Returns: + KSUID instance + """ + if len(ksuid_str) != _BASE36_STRING_LENGTH: + raise ValueError(f"Base36 KSUID string must be exactly {_BASE36_STRING_LENGTH} characters") + + decoded_bytes = _base36_decode(ksuid_str) + return cls.from_bytes(decoded_bytes) + + def to_base36(self) -> str: + """Return a lowercase base36-encoded string (31 characters).""" + return _base36_encode(self._bytes) + @classmethod def from_bytes(cls, data: bytes) -> 'KSUID': """ @@ -190,10 +222,10 @@ def _base62_encode(data: bytes) -> str: num, remainder = divmod(num, BASE62_BASE) result.append(BASE62_ALPHABET[remainder]) - # Pad to 27 characters for KSUID + # Pad to fixed width for KSUID result.reverse() encoded = ''.join(result) - return encoded.zfill(27) + return encoded.zfill(_BASE62_STRING_LENGTH) _BASE62_LOOKUP = {c: i for i, c in enumerate(BASE62_ALPHABET)} @@ -221,6 +253,46 @@ def _base62_decode(s: str) -> bytes: return num.to_bytes(TOTAL_LENGTH, 'big') +# --- Base36 (lowercase) encoding --------------------------------------------------- + +_BASE36_LOOKUP = {c: i for i, c in enumerate(BASE36_ALPHABET)} + + +def _base36_encode(data: bytes) -> str: + """Encode bytes to lowercase base36 string.""" + if not data: + return "" + + num = int.from_bytes(data, 'big') + + result = [] + while num > 0: + num, remainder = divmod(num, BASE36_BASE) + result.append(BASE36_ALPHABET[remainder]) + + result.reverse() + encoded = ''.join(result) + return encoded.zfill(_BASE36_STRING_LENGTH) + + +def _base36_decode(s: str) -> bytes: + """Decode lowercase base36 string to bytes.""" + if not s: + return b"" + + num = 0 + for char in s: + val = _BASE36_LOOKUP.get(char) + if val is None: + raise ValueError(f"Invalid base36 character: {char!r}") + num = num * BASE36_BASE + val + + if num > _MAX_ENCODED: + raise ValueError("Base36 value exceeds maximum for KSUID") + + return num.to_bytes(TOTAL_LENGTH, 'big') + + # Convenience functions def generate() -> KSUID: """Generate a new KSUID.""" @@ -241,11 +313,40 @@ def generate_token() -> str: return _base62_encode(secrets.token_bytes(TOTAL_LENGTH)) +def generate_lowercase() -> str: + """Generate a new KSUID and return it as a lowercase base36 string. + + The returned 31-character string uses only ``0-9a-z`` and preserves + the KSUID's timestamp + random-payload structure (sortable). + + Returns: + A 31-character lowercase base36 string. + """ + return KSUID().to_base36() + + +def generate_token_lowercase() -> str: + """Generate a cryptographically secure opaque token as a lowercase string. + + Uses 20 bytes (160 bits) of pure random data (no timestamp) encoded + in base36 (``0-9a-z`` only). + + Returns: + A 31-character lowercase base36 string with 160 bits of entropy. + """ + return _base36_encode(secrets.token_bytes(TOTAL_LENGTH)) + + def from_string(ksuid_str: str) -> KSUID: """Create a KSUID from its string representation.""" return KSUID.from_string(ksuid_str) +def from_base36(ksuid_str: str) -> KSUID: + """Create a KSUID from a lowercase base36 string representation.""" + return KSUID.from_base36(ksuid_str) + + def from_bytes(data: bytes) -> KSUID: """Create a KSUID from raw bytes.""" - return KSUID.from_bytes(data) \ No newline at end of file + return KSUID.from_bytes(data) \ No newline at end of file diff --git a/test_ksuid.py b/test_ksuid.py index 157e088..946f4db 100644 --- a/test_ksuid.py +++ b/test_ksuid.py @@ -9,7 +9,11 @@ import pytest from datetime import datetime, timezone -from __init__ import KSUID, generate, generate_token, from_string, from_bytes, EPOCH +from __init__ import ( + KSUID, generate, generate_lowercase, generate_token, generate_token_lowercase, + from_string, from_base36, from_bytes, EPOCH, + _BASE36_STRING_LENGTH, +) class TestKSUID: @@ -342,6 +346,105 @@ def gen_batch(_): assert len(set(all_tokens)) == len(all_tokens) +class TestLowercase: + """Test base36 lowercase encoding/decoding.""" + + def test_generate_lowercase_length(self): + """Lowercase KSUID must be exactly 31 characters.""" + s = generate_lowercase() + assert len(s) == _BASE36_STRING_LENGTH + + def test_generate_lowercase_charset(self): + """Lowercase KSUID must contain only 0-9a-z.""" + valid = set("0123456789abcdefghijklmnopqrstuvwxyz") + s = generate_lowercase() + assert all(c in valid for c in s), f"invalid chars in {s!r}" + + def test_generate_lowercase_has_no_uppercase(self): + """Must not contain any uppercase letter.""" + for _ in range(50): + s = generate_lowercase() + assert s == s.lower(), f"uppercase found in {s!r}" + + def test_lowercase_round_trip(self): + """KSUID -> to_base36 -> from_base36 must preserve identity.""" + ksuid = KSUID() + b36 = ksuid.to_base36() + restored = KSUID.from_base36(b36) + assert ksuid == restored + assert ksuid.timestamp == restored.timestamp + assert ksuid.payload == restored.payload + + def test_from_base36_convenience(self): + """Module-level from_base36() must work like KSUID.from_base36().""" + ksuid = generate() + b36 = ksuid.to_base36() + assert from_base36(b36) == ksuid + + def test_from_base36_invalid_length(self): + """Wrong-length base36 string must raise ValueError.""" + with pytest.raises(ValueError, match="exactly 31 characters"): + KSUID.from_base36("abc") + with pytest.raises(ValueError, match="exactly 31 characters"): + KSUID.from_base36("a" * 40) + + def test_from_base36_invalid_chars(self): + """Uppercase or special chars must be rejected.""" + with pytest.raises(ValueError, match="Invalid base36 character"): + KSUID.from_base36("A" * 31) + with pytest.raises(ValueError, match="Invalid base36 character"): + KSUID.from_base36("!" * 31) + + def test_from_base36_overflow(self): + """Max base36 31-char string must raise if it exceeds 20-byte max.""" + with pytest.raises(ValueError, match="exceeds maximum"): + KSUID.from_base36("z" * 31) + + def test_lowercase_sortability(self): + """Base36 strings of KSUIDs with increasing timestamps must sort.""" + ts1, ts2, ts3 = 1609459200, 1609459201, 1609459202 + payload = b'\x00' * 16 + s1 = KSUID(timestamp=ts1, payload=payload).to_base36() + s2 = KSUID(timestamp=ts2, payload=payload).to_base36() + s3 = KSUID(timestamp=ts3, payload=payload).to_base36() + assert s1 < s2 < s3 + + def test_generate_lowercase_uniqueness(self): + """100 lowercase KSUIDs must all be distinct.""" + ids = {generate_lowercase() for _ in range(100)} + assert len(ids) == 100 + + def test_generate_token_lowercase_length(self): + """Lowercase token must be exactly 31 characters.""" + assert len(generate_token_lowercase()) == _BASE36_STRING_LENGTH + + def test_generate_token_lowercase_charset(self): + """Lowercase token must contain only 0-9a-z.""" + valid = set("0123456789abcdefghijklmnopqrstuvwxyz") + token = generate_token_lowercase() + assert all(c in valid for c in token) + + def test_generate_token_lowercase_uniqueness(self): + """100 lowercase tokens must all be distinct.""" + tokens = {generate_token_lowercase() for _ in range(100)} + assert len(tokens) == 100 + + def test_zero_value_base36_round_trip(self): + """All-zero KSUID must encode to 31 '0' chars in base36.""" + ksuid = KSUID(timestamp=EPOCH, payload=b'\x00' * 16) + s = ksuid.to_base36() + assert len(s) == _BASE36_STRING_LENGTH + assert s == '0' * _BASE36_STRING_LENGTH + assert KSUID.from_base36(s) == ksuid + + def test_max_value_base36_round_trip(self): + """Max-timestamp, max-payload KSUID must round-trip via base36.""" + ksuid = KSUID(timestamp=EPOCH + 2**32 - 1, payload=b'\xff' * 16) + s = ksuid.to_base36() + assert len(s) == _BASE36_STRING_LENGTH + assert KSUID.from_base36(s) == ksuid + + class TestEdgeCases: """Edge cases for encoding / decoding.""" @@ -379,11 +482,10 @@ def test_max_timestamp_round_trip(self): assert ksuid1 == ksuid2 print("Round-trip test passed!") - # Test sortability - import time - time.sleep(0.001) # Ensure different timestamp - ksuid3 = generate() - assert ksuid1 < ksuid3 + # Test sortability (use explicit timestamps to avoid flaky 1ms sleep) + earlier = KSUID(timestamp=1609459200, payload=b'\x00' * 16) + later = KSUID(timestamp=1609459201, payload=b'\x00' * 16) + assert earlier < later print("Sortability test passed!") print("\nAll basic tests passed! Run with pytest for comprehensive testing.") \ No newline at end of file From 5eac15ae22e0a96825eb2e6a4ef63c00e8a26ec9 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Feb 2026 18:49:16 +0000 Subject: [PATCH 2/3] Rename __init__.py to ksuid.py, fix lint/Windows, update README - Rename __init__.py to ksuid.py for cross-platform imports (fixes Windows test failure where `from __init__ import` is invalid) - Remove sys.path.insert hacks from all files - Apply black formatting across all files - Fix all flake8 issues (unused imports, line length) - Update README to document full API: generate_lowercase(), generate_token(), generate_token_lowercase(), to_base36()/from_base36(), secure tokens, thread safety, __slots__, Python 3.9+ support https://claude.ai/code/session_01PVPVUNWhpxVa3xbDDBnwp2 --- README.md | 448 ++++++++++++---------------------------- benchmark.py | 83 ++++---- cli.py | 95 +++++---- example.py | 91 ++++---- __init__.py => ksuid.py | 113 +++++----- prefixed_examples.py | 265 ++++++++++++------------ test_ksuid.py | 188 +++++++++-------- 7 files changed, 567 insertions(+), 716 deletions(-) rename __init__.py => ksuid.py (86%) diff --git a/README.md b/README.md index ed21786..4f393ba 100644 --- a/README.md +++ b/README.md @@ -1,11 +1,9 @@ # KSUID - K-Sortable Unique Identifier -[![PyPI version](https://badge.fury.io/py/ksuid-python.svg)](https://pypi.org/project/ksuid-python/) [![Python Version](https://img.shields.io/pypi/pyversions/ksuid-python.svg)](https://pypi.org/project/ksuid-python/) [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) -[![Downloads](https://pepy.tech/badge/ksuid-python)](https://pepy.tech/project/ksuid-python) -A Python implementation of [KSUID](https://github.com/segmentio/ksuid) (K-Sortable Unique Identifier) for Python 3.13+. +A Python implementation of [KSUID](https://github.com/segmentio/ksuid) (K-Sortable Unique Identifier) for Python 3.9+. ## What is a KSUID? @@ -14,9 +12,12 @@ A KSUID is a globally unique identifier similar to a UUID, but with better prope - **Sortable**: KSUIDs are naturally sortable by creation time - **Compact**: 27 characters when base62-encoded (vs 36 for UUID) - **URL-safe**: Uses base62 encoding (no special characters) -- **Collision-resistant**: 128 bits of randomness per millisecond +- **Collision-resistant**: 128 bits of randomness per second - **Time-based**: Encodes creation timestamp for easy debugging - **Prefix-friendly**: Can be prefixed for type identification (like Stripe's API keys) +- **Lowercase option**: Base36 encoding for case-insensitive contexts (31 characters) +- **Secure tokens**: Generate timestamp-free tokens with 160 bits of entropy +- **Thread-safe**: Safe for concurrent use across multiple threads ## Format @@ -24,94 +25,60 @@ A KSUID is a 20-byte identifier consisting of: - **4 bytes**: Timestamp (seconds since KSUID epoch: 2014-05-13 16:53:20 UTC) - **16 bytes**: Random payload -When base62-encoded, it becomes a 27-character string like: `2StGMtcWzRJ8qZqQjbJjGdTkVfv` +| Encoding | Characters | Alphabet | Use case | +|----------|-----------|----------|----------| +| Base62 (default) | 27 | `0-9A-Za-z` | Standard, compact | +| Base36 (lowercase) | 31 | `0-9a-z` | Case-insensitive systems | -## Real-World Usage Examples - -Many successful companies use KSUID-style identifiers with prefixes for better developer experience: - -### Stripe-Style Prefixed IDs -```python -from ksuid import generate - -# Payment Intent: pi_1A2B3C... -payment_intent = f"pi_{generate()}" - -# Customer: cus_1A2B3C... -customer = f"cus_{generate()}" - -# Charge: ch_1A2B3C... -charge = f"ch_{generate()}" -``` - -### GitHub-Style IDs -```python -# Repository: repo_1A2B3C... -repository = f"repo_{generate()}" - -# Issue: issue_1A2B3C... -issue = f"issue_{generate()}" - -# Pull Request: pr_1A2B3C... -pull_request = f"pr_{generate()}" -``` +## Quick Start -### Database Entity IDs ```python -# User: user_1A2B3C... -user_id = f"user_{generate()}" +from ksuid import KSUID, generate, generate_lowercase -# Order: order_1A2B3C... -order_id = f"order_{generate()}" - -# Product: prod_1A2B3C... -product_id = f"prod_{generate()}" -``` +# Generate a new KSUID (base62, mixed-case) +ksuid = generate() +print(ksuid) # 2StGMtcWzRJ8qZqQjbJjGdTkVfv -### Benefits of Prefixed KSUIDs +# Generate a lowercase KSUID (base36) +lower_id = generate_lowercase() +print(lower_id) # 0c7de9014xkr8gqp3n7ewbz5jhr -1. **Type Safety**: Immediately identify the entity type -2. **Debugging**: Easier to trace issues in logs -3. **API Design**: Self-documenting API endpoints -4. **Database Queries**: Faster filtering by prefix -5. **Developer Experience**: Clear, readable identifiers +# Create from string +ksuid2 = KSUID.from_string("2StGMtcWzRJ8qZqQjbJjGdTkVfv") -## Installation +# Convert between formats +ksuid = KSUID() +str(ksuid) # Base62: "2StGMtcWzRJ8qZqQjbJjGdTkVfv" (27 chars) +ksuid.to_base36() # Base36: "0c7de9014xkr8gqp3n7ewbz5jhr" (31 chars) -Install from [PyPI](https://pypi.org/project/ksuid-python/): +# KSUIDs are sortable +ksuid_a = KSUID(timestamp=1609459200) +ksuid_b = KSUID(timestamp=1609459201) +assert ksuid_a < ksuid_b # True! -```bash -pip install ksuid-python +# Access timestamp and payload +print(ksuid.datetime) # 2025-01-17 10:30:45+00:00 +print(ksuid.timestamp) # 1737108645 +print(len(ksuid.payload)) # 16 bytes ``` -**Note:** The package name is `ksuid-python`, but you import it as `ksuid`: - -```python -from ksuid import generate # Import name is 'ksuid' -``` +## Secure Tokens -## Quick Start +For API keys, session tokens, and other security-sensitive values, use +`generate_token()` or `generate_token_lowercase()`. These use 160 bits of +`secrets.token_bytes` randomness with **no embedded timestamp**, so creation +time cannot be leaked. ```python -from ksuid import KSUID, generate - -# Generate a new KSUID -ksuid = generate() -print(ksuid) # 2StGMtcWzRJ8qZqQjbJjGdTkVfv +from ksuid import generate_token, generate_token_lowercase -# Create from string -ksuid2 = KSUID.from_string('2StGMtcWzRJ8qZqQjbJjGdTkVfv') - -# KSUIDs are sortable -ksuid1 = generate() -time.sleep(0.001) -ksuid2 = generate() -assert ksuid1 < ksuid2 # True! +# Mixed-case token (27 chars, base62, 160-bit entropy) +api_key = f"sk_{generate_token()}" +# sk_7kQ9xLm3RtN5vW8yBzCdEfGhJ -# Access timestamp and payload -print(ksuid.datetime) # 2025-01-17 10:30:45+00:00 -print(ksuid.timestamp) # 1737108645 -print(len(ksuid.payload)) # 16 bytes +# Lowercase token (31 chars, base36, 160-bit entropy) +session = f"sess_{generate_token_lowercase()}" +# sess_4f8a2bc90d1e3f5g6h7i8j9k0lm ``` ## API Reference @@ -129,90 +96,70 @@ KSUID(timestamp=None, payload=None) #### Class Methods -```python -KSUID.from_string(ksuid_str: str) -> KSUID -``` -Create a KSUID from its base62 string representation. - -```python -KSUID.from_bytes(data: bytes) -> KSUID -``` -Create a KSUID from raw 20-byte data. +| Method | Description | +|--------|-------------| +| `KSUID.from_string(s)` | Create from 27-char base62 string | +| `KSUID.from_base36(s)` | Create from 31-char lowercase base36 string | +| `KSUID.from_bytes(data)` | Create from raw 20-byte data | #### Properties -- `timestamp`: Unix timestamp (int) -- `datetime`: Python datetime object (UTC) -- `payload`: 16-byte random payload (bytes) -- `bytes`: Raw 20-byte KSUID data (bytes) +| Property | Type | Description | +|----------|------|-------------| +| `timestamp` | `int` | Unix timestamp | +| `datetime` | `datetime` | UTC datetime object | +| `payload` | `bytes` | 16-byte random payload | +| `bytes` | `bytes` | Raw 20-byte KSUID data | #### Methods -- `__str__()`: Returns base62-encoded string representation -- `__repr__()`: Returns developer-friendly representation -- Comparison operators: `<`, `<=`, `>`, `>=`, `==`, `!=` -- `__hash__()`: Makes KSUIDs hashable (usable in sets/dicts) +| Method | Description | +|--------|-------------| +| `__str__()` | Base62-encoded string (27 chars) | +| `to_base36()` | Base36-encoded lowercase string (31 chars) | +| `__repr__()` | Developer-friendly representation | +| `__hash__()` | Hashable (usable in sets/dicts) | +| `<`, `<=`, `>`, `>=`, `==`, `!=` | Sortable comparison | ### Convenience Functions -```python -generate() -> KSUID -``` -Generate a new KSUID with current timestamp. - -```python -from_string(ksuid_str: str) -> KSUID -``` -Create KSUID from string (alias for `KSUID.from_string`). - -```python -from_bytes(data: bytes) -> KSUID -``` -Create KSUID from bytes (alias for `KSUID.from_bytes`). +| Function | Returns | Description | +|----------|---------|-------------| +| `generate()` | `KSUID` | New KSUID with current timestamp | +| `generate_lowercase()` | `str` | 31-char lowercase base36 KSUID (sortable) | +| `generate_token()` | `str` | 27-char base62 secure token (no timestamp) | +| `generate_token_lowercase()` | `str` | 31-char base36 secure token (no timestamp) | +| `from_string(s)` | `KSUID` | Parse base62 string | +| `from_base36(s)` | `KSUID` | Parse base36 string | +| `from_bytes(data)` | `KSUID` | Parse raw bytes | ## Examples -### Basic Usage +### Prefixed IDs (Stripe-Style) ```python -from ksuid import KSUID, generate -import time +from ksuid import generate, generate_lowercase -# Generate KSUIDs -ksuid1 = generate() -time.sleep(0.001) -ksuid2 = generate() +# Mixed-case prefixed IDs +user_id = f"user_{generate()}" # user_2StGMtcWzRJ8qZqQjbJjGdTkVfv +payment_id = f"pi_{generate()}" # pi_2StGMtcWzRJ8qZqQjbJjGdTkVfv -print(f"KSUID 1: {ksuid1}") -print(f"KSUID 2: {ksuid2}") -print(f"KSUID 1 < KSUID 2: {ksuid1 < ksuid2}") # True +# Lowercase prefixed IDs +user_id = f"user_{generate_lowercase()}" # user_0c7de9014xkr8gqp3n7ewbz5jhr +order_id = f"ord_{generate_lowercase()}" # ord_0c7de9014xkr8gqp3n7ewbz5jhr ``` -### Prefixed IDs (Stripe-Style) +### Secure API Keys and Session Tokens ```python -from ksuid import generate - -def create_prefixed_id(prefix: str) -> str: - """Create a prefixed ID like Stripe's API keys.""" - return f"{prefix}_{generate()}" - -# Create different entity types -user_id = create_prefixed_id("user") # user_2StGMtcWzRJ8qZqQjbJjGdTkVfv -payment_id = create_prefixed_id("pi") # pi_2StGMtcWzRJ8qZqQjbJjGdTkVfv -customer_id = create_prefixed_id("cus") # cus_2StGMtcWzRJ8qZqQjbJjGdTkVfv +from ksuid import generate_token, generate_token_lowercase -print(f"User ID: {user_id}") -print(f"Payment ID: {payment_id}") -print(f"Customer ID: {customer_id}") +# API keys (no timestamp leakage, 160-bit entropy) +secret_key = f"sk_{generate_token()}" +public_key = f"pk_{generate_token()}" -# Extract KSUID from prefixed ID -def extract_ksuid(prefixed_id: str) -> str: - """Extract KSUID from prefixed ID.""" - return prefixed_id.split('_', 1)[1] - -ksuid_part = extract_ksuid(user_id) -print(f"Extracted KSUID: {ksuid_part}") +# Lowercase session tokens +session_id = f"sess_{generate_token_lowercase()}" ``` ### Custom Timestamp and Payload @@ -233,14 +180,10 @@ ksuid = KSUID(payload=payload) ### Sorting and Comparison ```python -from ksuid import generate -import time +from ksuid import KSUID, generate -# Generate multiple KSUIDs -ksuids = [] -for i in range(5): - ksuids.append(generate()) - time.sleep(0.001) +# Generate KSUIDs with different timestamps +ksuids = [KSUID(timestamp=1609459200 + i) for i in range(5)] # They're naturally sorted by creation time sorted_ksuids = sorted(ksuids) @@ -251,111 +194,44 @@ ksuid_set = set(ksuids) ksuid_dict = {k: f"value_{i}" for i, k in enumerate(ksuids)} ``` +### Converting Between Formats + +```python +from ksuid import KSUID, from_base36 + +# Start with a KSUID +ksuid = KSUID() + +# Get different representations +b62 = str(ksuid) # Base62: 27 chars +b36 = ksuid.to_base36() # Base36: 31 chars, lowercase +raw = ksuid.bytes # Raw: 20 bytes + +# Recreate from any representation +assert KSUID.from_string(b62) == ksuid +assert KSUID.from_base36(b36) == ksuid +assert KSUID.from_bytes(raw) == ksuid +``` + ### Database Usage ```python -from ksuid import generate +from ksuid import generate_lowercase import sqlite3 -# Create table with KSUID primary key conn = sqlite3.connect(':memory:') conn.execute(''' CREATE TABLE users ( id TEXT PRIMARY KEY, - name TEXT, - created_at DATETIME + name TEXT ) ''') -# Insert records with KSUID -ksuid = generate() +user_id = f"user_{generate_lowercase()}" conn.execute( - 'INSERT INTO users (id, name, created_at) VALUES (?, ?, ?)', - (str(ksuid), 'John Doe', ksuid.datetime) + 'INSERT INTO users (id, name) VALUES (?, ?)', + (user_id, 'John Doe') ) - -# Query by KSUID -cursor = conn.execute('SELECT * FROM users WHERE id = ?', (str(ksuid),)) -print(cursor.fetchone()) -``` - -### Production API Example (Flask) - -```python -from flask import Flask, jsonify, request -from ksuid import generate -import sqlite3 - -app = Flask(__name__) - -def create_prefixed_id(prefix: str) -> str: - return f"{prefix}_{generate()}" - -@app.route('/api/users', methods=['POST']) -def create_user(): - data = request.json - user_id = create_prefixed_id("user") - - # Store in database - conn = sqlite3.connect('app.db') - conn.execute( - 'INSERT INTO users (id, name, email) VALUES (?, ?, ?)', - (user_id, data['name'], data['email']) - ) - conn.commit() - conn.close() - - return jsonify({ - 'id': user_id, - 'name': data['name'], - 'email': data['email'] - }), 201 - -@app.route('/api/users/') -def get_user(user_id): - # Validate prefix - if not user_id.startswith('user_'): - return jsonify({'error': 'Invalid user ID format'}), 400 - - conn = sqlite3.connect('app.db') - cursor = conn.execute('SELECT * FROM users WHERE id = ?', (user_id,)) - user = cursor.fetchone() - conn.close() - - if not user: - return jsonify({'error': 'User not found'}), 404 - - return jsonify({ - 'id': user[0], - 'name': user[1], - 'email': user[2] - }) - -# Example usage: -# POST /api/users -> {"id": "user_2StGMtcWzRJ8qZqQjbJjGdTkVfv", ...} -# GET /api/users/user_2StGMtcWzRJ8qZqQjbJjGdTkVfv -> User details -``` - -### Converting Between Formats - -```python -from ksuid import KSUID - -# Start with a KSUID -ksuid = KSUID() - -# Get different representations -string_repr = str(ksuid) # Base62 string -bytes_repr = ksuid.bytes # Raw bytes -timestamp = ksuid.timestamp # Unix timestamp -datetime_obj = ksuid.datetime # Python datetime - -# Recreate from representations -ksuid_from_string = KSUID.from_string(string_repr) -ksuid_from_bytes = KSUID.from_bytes(bytes_repr) - -# All should be equal -assert ksuid == ksuid_from_string == ksuid_from_bytes ``` ## Performance @@ -365,87 +241,31 @@ KSUIDs are designed to be fast and efficient: - **Generation**: ~1-2 microseconds per KSUID - **Parsing**: ~500 nanoseconds from string - **Comparison**: ~100 nanoseconds -- **Memory**: 20 bytes per KSUID + Python object overhead +- **Memory**: Optimized with `__slots__` (~48 bytes per instance) ## Comparison with UUIDs -| Feature | KSUID | UUID v4 | UUID v1 | Stripe IDs | -|---------|-------|---------|---------|------------| -| Length | 27 chars | 36 chars | 36 chars | 24-28 chars | -| Sortable | ✅ Yes | ❌ No | ⚠️ Partially | ❌ No | -| URL-safe | ✅ Yes | ❌ No (hyphens) | ❌ No (hyphens) | ✅ Yes | -| Timestamp | ✅ Readable | ❌ No | ✅ But complex | ❌ No | -| Collision resistance | ✅ High | ✅ High | ✅ High | ✅ High | -| Monotonic | ✅ Yes | ❌ No | ⚠️ Partially | ❌ No | -| Prefix support | ✅ Natural | ❌ Awkward | ❌ Awkward | ✅ Built-in | -| Developer UX | ✅ Excellent | ⚠️ Good | ⚠️ Good | ✅ Excellent | - -## Industry Adoption & Best Practices - -### Companies Using KSUID-Style IDs - -Many successful companies use sortable, prefixed identifiers: - -- **Stripe**: `pi_1A2B3C...`, `cus_1A2B3C...`, `ch_1A2B3C...` -- **GitHub**: Repository and issue IDs with chronological ordering -- **Slack**: Channel and message IDs for efficient sorting -- **Discord**: Snowflake IDs (similar concept with timestamps) -- **Twitter**: Tweet IDs (chronologically sortable) - -### Prefix Naming Conventions - -Common patterns for prefixes: - -```python -# Entity types (3-4 chars) -user_id = f"user_{generate()}" # Users -prod_id = f"prod_{generate()}" # Products -ord_id = f"ord_{generate()}" # Orders - -# Action types (2-3 chars) -payment_id = f"pi_{generate()}" # Payment Intent (Stripe style) -charge_id = f"ch_{generate()}" # Charge -refund_id = f"re_{generate()}" # Refund - -# Short codes (2-3 chars) -api_key = f"sk_{generate()}" # Secret Key -pub_key = f"pk_{generate()}" # Public Key -token = f"tok_{generate()}" # Token -``` - -### Database Design Tips - -```sql --- Index on prefix for fast filtering -CREATE INDEX idx_users_by_type ON transactions(id) WHERE id LIKE 'user_%'; - --- Partial indexes for different entity types -CREATE INDEX idx_payments ON transactions(id) WHERE id LIKE 'pi_%'; -CREATE INDEX idx_refunds ON transactions(id) WHERE id LIKE 're_%'; -``` - -### API Design Patterns - -```python -# RESTful endpoints with typed IDs -GET /api/users/user_2StGMtcWzRJ8qZqQjbJjGdTkVfv -GET /api/payments/pi_2StGMtcWzRJ8qZqQjbJjGdTkVfv -GET /api/orders/ord_2StGMtcWzRJ8qZqQjbJjGdTkVfv - -# Validation middleware -def validate_entity_id(entity_type, entity_id): - if not entity_id.startswith(f"{entity_type}_"): - raise ValueError(f"Invalid {entity_type} ID format") - return entity_id.split('_', 1)[1] # Return KSUID part -``` +| Feature | KSUID | UUID v4 | Stripe IDs | +|---------|-------|---------|------------| +| Length | 27 chars (base62) / 31 chars (base36) | 36 chars | 24-28 chars | +| Sortable | Yes | No | No | +| URL-safe | Yes | No (hyphens) | Yes | +| Timestamp | Readable | No | No | +| Collision resistance | High (128 bits) | High (122 bits) | High | +| Lowercase option | Yes (base36) | Yes (already) | No | +| Secure tokens | Yes (`generate_token`) | No | No | +| Thread-safe | Yes | Yes | Yes | ## Thread Safety -The KSUID library is thread-safe. Multiple threads can generate KSUIDs concurrently without coordination. +The KSUID library is fully thread-safe. All functions use only thread-safe +primitives (`os.urandom`, `secrets.token_bytes`, `time.time`) and KSUID +instances are immutable after construction. Safe even under free-threaded +Python 3.13+ (no-GIL). ## Requirements -- Python 3.13 or later +- Python 3.9 or later - No external dependencies ## Development @@ -456,19 +276,19 @@ git clone https://github.com/tonyzorin/ksuid-python.git cd ksuid-python # Install development dependencies -pip install -e ".[dev]" +pip install pytest pytest-cov black flake8 mypy # Run tests -pytest +pytest test_ksuid.py -v # Run tests with coverage -pytest --cov=ksuid +pytest test_ksuid.py --cov=ksuid # Format code black . -# Type checking -mypy ksuid/ +# Lint +flake8 . --max-line-length=88 --extend-ignore=E203,W503 ``` ## License @@ -482,4 +302,4 @@ Contributions are welcome! Please feel free to submit a Pull Request. ## References - [Original KSUID specification](https://github.com/segmentio/ksuid) -- [KSUID in other languages](https://github.com/segmentio/ksuid#other-implementations) +- [KSUID in other languages](https://github.com/segmentio/ksuid#other-implementations) diff --git a/benchmark.py b/benchmark.py index aa8ebc9..64c9f08 100644 --- a/benchmark.py +++ b/benchmark.py @@ -6,54 +6,53 @@ """ import time -import sys -import os -sys.path.insert(0, os.path.dirname(__file__)) -from __init__ import KSUID, generate, from_string, from_bytes +from ksuid import generate, from_string, from_bytes def benchmark_generation(count=100000): """Benchmark KSUID generation.""" print(f"Benchmarking KSUID generation ({count:,} iterations)...") - + start_time = time.perf_counter() ksuids = [generate() for _ in range(count)] end_time = time.perf_counter() - + total_time = end_time - start_time rate = count / total_time avg_time_us = (total_time / count) * 1_000_000 - + print(f" Total time: {total_time:.4f} seconds") print(f" Rate: {rate:,.0f} KSUIDs/second") print(f" Average time: {avg_time_us:.2f} microseconds per KSUID") - + # Verify uniqueness unique_count = len(set(ksuids)) collision_rate = (count - unique_count) / count * 100 - print(f" Uniqueness: {unique_count:,} / {count:,} ({collision_rate:.6f}% collisions)") + print( + f" Uniqueness: {unique_count:,} / {count:,} ({collision_rate:.6f}% collisions)" + ) print() - + return ksuids def benchmark_string_parsing(ksuids, iterations=10000): """Benchmark string parsing.""" print(f"Benchmarking string parsing ({iterations:,} iterations)...") - + # Use a subset of KSUIDs for parsing test_strings = [str(ksuid) for ksuid in ksuids[:iterations]] - + start_time = time.perf_counter() for ksuid_str in test_strings: from_string(ksuid_str) end_time = time.perf_counter() - + total_time = end_time - start_time rate = iterations / total_time avg_time_ns = (total_time / iterations) * 1_000_000_000 - + print(f" Total time: {total_time:.4f} seconds") print(f" Rate: {rate:,.0f} parses/second") print(f" Average time: {avg_time_ns:.0f} nanoseconds per parse") @@ -63,19 +62,19 @@ def benchmark_string_parsing(ksuids, iterations=10000): def benchmark_bytes_parsing(ksuids, iterations=10000): """Benchmark bytes parsing.""" print(f"Benchmarking bytes parsing ({iterations:,} iterations)...") - + # Use a subset of KSUIDs for parsing test_bytes = [ksuid.bytes for ksuid in ksuids[:iterations]] - + start_time = time.perf_counter() for ksuid_bytes in test_bytes: from_bytes(ksuid_bytes) end_time = time.perf_counter() - + total_time = end_time - start_time rate = iterations / total_time avg_time_ns = (total_time / iterations) * 1_000_000_000 - + print(f" Total time: {total_time:.4f} seconds") print(f" Rate: {rate:,.0f} parses/second") print(f" Average time: {avg_time_ns:.0f} nanoseconds per parse") @@ -85,19 +84,21 @@ def benchmark_bytes_parsing(ksuids, iterations=10000): def benchmark_comparison(ksuids, iterations=100000): """Benchmark KSUID comparison.""" print(f"Benchmarking KSUID comparison ({iterations:,} iterations)...") - + # Create pairs for comparison - pairs = [(ksuids[i], ksuids[i+1]) for i in range(0, min(iterations, len(ksuids)-1))] - + pairs = [ + (ksuids[i], ksuids[i + 1]) for i in range(0, min(iterations, len(ksuids) - 1)) + ] + start_time = time.perf_counter() for ksuid1, ksuid2 in pairs: _ = ksuid1 < ksuid2 end_time = time.perf_counter() - + total_time = end_time - start_time rate = len(pairs) / total_time avg_time_ns = (total_time / len(pairs)) * 1_000_000_000 - + print(f" Total time: {total_time:.4f} seconds") print(f" Rate: {rate:,.0f} comparisons/second") print(f" Average time: {avg_time_ns:.0f} nanoseconds per comparison") @@ -107,26 +108,29 @@ def benchmark_comparison(ksuids, iterations=100000): def benchmark_sorting(ksuids, count=10000): """Benchmark KSUID sorting.""" print(f"Benchmarking KSUID sorting ({count:,} items)...") - + # Shuffle KSUIDs for sorting import random + test_ksuids = ksuids[:count].copy() random.shuffle(test_ksuids) - + start_time = time.perf_counter() sorted_ksuids = sorted(test_ksuids) end_time = time.perf_counter() - + total_time = end_time - start_time rate = count / total_time avg_time_us = (total_time / count) * 1_000_000 - + print(f" Total time: {total_time:.4f} seconds") print(f" Rate: {rate:,.0f} items/second") print(f" Average time: {avg_time_us:.2f} microseconds per item") - + # Verify sorting worked - is_sorted = all(sorted_ksuids[i] <= sorted_ksuids[i+1] for i in range(len(sorted_ksuids)-1)) + is_sorted = all( + sorted_ksuids[i] <= sorted_ksuids[i + 1] for i in range(len(sorted_ksuids) - 1) + ) print(f" Correctly sorted: {is_sorted}") print() @@ -134,22 +138,25 @@ def benchmark_sorting(ksuids, count=10000): def benchmark_memory_usage(count=10000): """Estimate memory usage of KSUIDs.""" print(f"Estimating memory usage ({count:,} KSUIDs)...") - + import sys - + # Measure memory of a single KSUID ksuid = generate() ksuid_size = sys.getsizeof(ksuid) string_size = sys.getsizeof(str(ksuid)) bytes_size = sys.getsizeof(ksuid.bytes) - + print(f" KSUID object size: {ksuid_size} bytes") print(f" String representation: {string_size} bytes") print(f" Raw bytes: {bytes_size} bytes") - + # Estimate total memory for collection estimated_total = count * (ksuid_size + 56) # +56 for dict/list overhead - print(f" Estimated total for {count:,} KSUIDs: {estimated_total:,} bytes ({estimated_total/1024/1024:.2f} MB)") + print( + f" Estimated total for {count:,} KSUIDs: " + f"{estimated_total:,} bytes ({estimated_total/1024/1024:.2f} MB)" + ) print() @@ -158,17 +165,17 @@ def main(): print("KSUID Performance Benchmark") print("=" * 50) print() - + # Generate KSUIDs for testing ksuids = benchmark_generation(100000) - + # Run various benchmarks benchmark_string_parsing(ksuids, 10000) benchmark_bytes_parsing(ksuids, 10000) benchmark_comparison(ksuids, 50000) benchmark_sorting(ksuids, 10000) benchmark_memory_usage(10000) - + print("=== Performance Summary ===") print("KSUID operations are highly optimized:") print("✅ Generation: ~300k+ KSUIDs/second") @@ -180,4 +187,4 @@ def main(): if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/cli.py b/cli.py index 094d23c..100ea0c 100644 --- a/cli.py +++ b/cli.py @@ -7,13 +7,10 @@ import argparse import sys -import os -sys.path.insert(0, os.path.dirname(__file__)) -from __init__ import KSUID, generate, from_string +from ksuid import generate, from_string from datetime import datetime - MAX_COUNT = 1_000_000 @@ -26,9 +23,7 @@ def _validate_count(value): if ivalue < 1: raise argparse.ArgumentTypeError("count must be at least 1") if ivalue > MAX_COUNT: - raise argparse.ArgumentTypeError( - f"count must be at most {MAX_COUNT:,}" - ) + raise argparse.ArgumentTypeError(f"count must be at most {MAX_COUNT:,}") return ivalue @@ -40,7 +35,7 @@ def cmd_generate(args): result = f"{args.prefix}_{ksuid}" else: result = str(ksuid) - + if args.verbose: if args.prefix: print(f"{result} -> {ksuid.datetime} (timestamp: {ksuid.timestamp})") @@ -54,18 +49,18 @@ def cmd_inspect(args): """Inspect a KSUID and show its components.""" try: ksuid = from_string(args.ksuid) - + print(f"KSUID: {ksuid}") print(f"Timestamp: {ksuid.timestamp}") print(f"Datetime: {ksuid.datetime}") print(f"Payload: {ksuid.payload.hex()}") print(f"Raw bytes: {ksuid.bytes.hex()}") - + # Calculate age now = datetime.now(ksuid.datetime.tzinfo) age = now - ksuid.datetime print(f"Age: {age}") - + except ValueError as e: print(f"Error: {e}", file=sys.stderr) sys.exit(1) @@ -76,14 +71,14 @@ def cmd_compare(args): try: ksuid1 = from_string(args.ksuid1) ksuid2 = from_string(args.ksuid2) - + print(f"KSUID 1: {ksuid1}") print(f" Timestamp: {ksuid1.datetime}") print() print(f"KSUID 2: {ksuid2}") print(f" Timestamp: {ksuid2.datetime}") print() - + if ksuid1 == ksuid2: print("Result: KSUIDs are identical") elif ksuid1 < ksuid2: @@ -94,7 +89,7 @@ def cmd_compare(args): print("Result: KSUID 1 is newer than KSUID 2") time_diff = ksuid1.datetime - ksuid2.datetime print(f"Time difference: {time_diff}") - + except ValueError as e: print(f"Error: {e}", file=sys.stderr) sys.exit(1) @@ -103,23 +98,26 @@ def cmd_compare(args): def cmd_benchmark(args): """Run a simple benchmark.""" import time - + print(f"Benchmarking KSUID generation ({args.count:,} iterations)...") - + start_time = time.perf_counter() ksuids = [generate() for _ in range(args.count)] end_time = time.perf_counter() - + total_time = end_time - start_time rate = args.count / total_time - + print(f"Generated {args.count:,} KSUIDs in {total_time:.4f} seconds") print(f"Rate: {rate:,.0f} KSUIDs/second") - + # Check uniqueness unique_count = len(set(ksuids)) collision_rate = (args.count - unique_count) / args.count * 100 - print(f"Uniqueness: {unique_count:,} / {args.count:,} ({collision_rate:.6f}% collisions)") + print( + f"Uniqueness: {unique_count:,} / {args.count:,} " + f"({collision_rate:.6f}% collisions)" + ) def main(): @@ -137,62 +135,63 @@ def main(): %(prog)s inspect 2StGMtcWzRJ8qZqQjbJjGdTkVfv # Inspect a KSUID %(prog)s compare KSUID1 KSUID2 # Compare two KSUIDs %(prog)s benchmark -c 10000 # Benchmark generation - """ + """, ) - - subparsers = parser.add_subparsers(dest='command', help='Available commands') - + + subparsers = parser.add_subparsers(dest="command", help="Available commands") + # Generate command - gen_parser = subparsers.add_parser('generate', help='Generate KSUIDs') + gen_parser = subparsers.add_parser("generate", help="Generate KSUIDs") gen_parser.add_argument( - '-c', '--count', + "-c", + "--count", type=_validate_count, default=1, - help='Number of KSUIDs to generate (default: 1, max: 1,000,000)' + help="Number of KSUIDs to generate (default: 1, max: 1,000,000)", ) gen_parser.add_argument( - '-v', '--verbose', - action='store_true', - help='Show additional information' + "-v", "--verbose", action="store_true", help="Show additional information" ) gen_parser.add_argument( - '-p', '--prefix', - type=str, - help='Add prefix to KSUID (e.g., user, pi, cus for Stripe-style IDs)' + "-p", + "--prefix", + type=str, + help="Add prefix to KSUID (e.g., user, pi, cus for Stripe-style IDs)", ) gen_parser.set_defaults(func=cmd_generate) - + # Inspect command - inspect_parser = subparsers.add_parser('inspect', help='Inspect a KSUID') - inspect_parser.add_argument('ksuid', help='KSUID to inspect') + inspect_parser = subparsers.add_parser("inspect", help="Inspect a KSUID") + inspect_parser.add_argument("ksuid", help="KSUID to inspect") inspect_parser.set_defaults(func=cmd_inspect) - + # Compare command - compare_parser = subparsers.add_parser('compare', help='Compare two KSUIDs') - compare_parser.add_argument('ksuid1', help='First KSUID') - compare_parser.add_argument('ksuid2', help='Second KSUID') + compare_parser = subparsers.add_parser("compare", help="Compare two KSUIDs") + compare_parser.add_argument("ksuid1", help="First KSUID") + compare_parser.add_argument("ksuid2", help="Second KSUID") compare_parser.set_defaults(func=cmd_compare) - + # Benchmark command - bench_parser = subparsers.add_parser('benchmark', help='Run benchmark') + bench_parser = subparsers.add_parser("benchmark", help="Run benchmark") bench_parser.add_argument( - '-c', '--count', + "-c", + "--count", type=_validate_count, default=10000, - help='Number of KSUIDs to generate (default: 10000, max: 1,000,000)' + help="Number of KSUIDs to generate (default: 10000, max: 1,000,000)", ) bench_parser.set_defaults(func=cmd_benchmark) - + # Parse arguments args = parser.parse_args() - + if not args.command: parser.print_help() sys.exit(1) - + # Execute command args.func(args) if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/example.py b/example.py index 6b3991e..9fb0546 100644 --- a/example.py +++ b/example.py @@ -7,18 +7,15 @@ """ import time -import sys -import os -sys.path.insert(0, os.path.dirname(__file__)) from datetime import datetime, timezone -from __init__ import KSUID, generate, from_string +from ksuid import KSUID, generate, from_string def basic_usage(): """Demonstrate basic KSUID operations.""" print("=== Basic KSUID Usage ===") - + # Generate a new KSUID ksuid1 = generate() print(f"Generated KSUID: {ksuid1}") @@ -27,7 +24,7 @@ def basic_usage(): print(f"Unix timestamp: {ksuid1.timestamp}") print(f"Payload length: {len(ksuid1.payload)} bytes") print() - + # Create KSUID from string ksuid_str = str(ksuid1) ksuid2 = from_string(ksuid_str) @@ -39,22 +36,25 @@ def basic_usage(): def sortability_demo(): """Demonstrate KSUID sortability.""" print("=== KSUID Sortability Demo ===") - + # Generate KSUIDs with small time gaps ksuids = [] for i in range(5): ksuid = generate() ksuids.append(ksuid) - print(f"KSUID {i+1}: {ksuid} (created at {ksuid.datetime.strftime('%H:%M:%S.%f')})") + print( + f"KSUID {i+1}: {ksuid} " + f"(created at {ksuid.datetime.strftime('%H:%M:%S.%f')})" + ) time.sleep(0.001) # 1ms delay - + print("\nSorting KSUIDs...") sorted_ksuids = sorted(ksuids) - + print("Sorted order:") for i, ksuid in enumerate(sorted_ksuids): print(f" {i+1}. {ksuid}") - + # Verify they're in chronological order is_sorted = ksuids == sorted_ksuids print(f"\nAre they naturally sorted? {is_sorted}") @@ -64,21 +64,21 @@ def sortability_demo(): def custom_timestamp_demo(): """Demonstrate KSUIDs with custom timestamps.""" print("=== Custom Timestamp Demo ===") - + # Create KSUIDs for specific dates dates = [ datetime(2021, 1, 1, 0, 0, 0, tzinfo=timezone.utc), datetime(2022, 6, 15, 12, 30, 0, tzinfo=timezone.utc), datetime(2023, 12, 31, 23, 59, 59, tzinfo=timezone.utc), ] - + historical_ksuids = [] for date in dates: timestamp = int(date.timestamp()) ksuid = KSUID(timestamp=timestamp) historical_ksuids.append(ksuid) print(f"KSUID for {date}: {ksuid}") - + print("\nSorting historical KSUIDs:") for ksuid in sorted(historical_ksuids): print(f" {ksuid} -> {ksuid.datetime}") @@ -88,20 +88,20 @@ def custom_timestamp_demo(): def performance_demo(): """Demonstrate KSUID performance.""" print("=== Performance Demo ===") - + # Time KSUID generation start_time = time.perf_counter() count = 10000 - + ksuids = [generate() for _ in range(count)] - + end_time = time.perf_counter() total_time = end_time - start_time - + print(f"Generated {count:,} KSUIDs in {total_time:.4f} seconds") print(f"Rate: {count/total_time:,.0f} KSUIDs/second") print(f"Average time per KSUID: {(total_time/count)*1_000_000:.2f} microseconds") - + # Verify uniqueness unique_ksuids = set(ksuids) print(f"Unique KSUIDs: {len(unique_ksuids):,} / {count:,}") @@ -112,43 +112,42 @@ def performance_demo(): def database_simulation(): """Simulate database usage with KSUIDs.""" print("=== Database Simulation ===") - + # Simulate user records with KSUID primary keys users = [] - + user_names = ["Alice", "Bob", "Charlie", "Diana", "Eve"] - + for name in user_names: user_id = generate() user = { - 'id': str(user_id), - 'name': name, - 'created_at': user_id.datetime, - 'ksuid_obj': user_id # Keep KSUID object for sorting + "id": str(user_id), + "name": name, + "created_at": user_id.datetime, + "ksuid_obj": user_id, # Keep KSUID object for sorting } users.append(user) time.sleep(0.001) # Simulate time between user creations - + print("Created users:") for user in users: - print(f" ID: {user['id']}, Name: {user['name']}, Created: {user['created_at']}") - + print( + f" ID: {user['id']}, Name: {user['name']}, Created: {user['created_at']}" + ) + # Sort by creation time using KSUID print("\nUsers sorted by creation time (using KSUID sorting):") - sorted_users = sorted(users, key=lambda u: u['ksuid_obj']) + sorted_users = sorted(users, key=lambda u: u["ksuid_obj"]) for user in sorted_users: print(f" {user['name']} -> {user['created_at'].strftime('%H:%M:%S.%f')}") - + # Demonstrate range queries print("\nSimulating range query (users created in last 10ms):") now = generate() cutoff_time = now.datetime.timestamp() - 0.01 # 10ms ago - - recent_users = [ - user for user in users - if user['ksuid_obj'].timestamp > cutoff_time - ] - + + recent_users = [user for user in users if user["ksuid_obj"].timestamp > cutoff_time] + print(f"Found {len(recent_users)} recent users") print() @@ -156,27 +155,27 @@ def database_simulation(): def format_conversion_demo(): """Demonstrate format conversions.""" print("=== Format Conversion Demo ===") - + ksuid = generate() - + print(f"Original KSUID: {ksuid}") print(f"String representation: {str(ksuid)}") print(f"Bytes representation: {ksuid.bytes.hex()}") print(f"Timestamp: {ksuid.timestamp}") print(f"Datetime: {ksuid.datetime}") print(f"Payload: {ksuid.payload.hex()}") - + # Round-trip conversions print("\nRound-trip conversions:") - + # String round-trip ksuid_from_string = from_string(str(ksuid)) print(f"From string: {ksuid == ksuid_from_string}") - + # Bytes round-trip ksuid_from_bytes = KSUID.from_bytes(ksuid.bytes) print(f"From bytes: {ksuid == ksuid_from_bytes}") - + # Timestamp + payload round-trip ksuid_reconstructed = KSUID(timestamp=ksuid.timestamp, payload=ksuid.payload) print(f"Reconstructed: {ksuid == ksuid_reconstructed}") @@ -188,14 +187,14 @@ def main(): print("KSUID Library Demonstration") print("=" * 50) print() - + basic_usage() sortability_demo() custom_timestamp_demo() performance_demo() database_simulation() format_conversion_demo() - + print("=== Summary ===") print("KSUIDs provide:") print("✅ Sortable unique identifiers") @@ -207,4 +206,4 @@ def main(): if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/__init__.py b/ksuid.py similarity index 86% rename from __init__.py rename to ksuid.py index 56503cf..3360885 100644 --- a/__init__.py +++ b/ksuid.py @@ -10,10 +10,10 @@ >>> ksuid = KSUID() >>> str(ksuid) '2StGMtcWzRJ8qZqQjbJjGdTkVfv' - + >>> # Create from string >>> ksuid2 = KSUID.from_string('2StGMtcWzRJ8qZqQjbJjGdTkVfv') - + >>> # Compare KSUIDs (they're sortable) >>> ksuid1 < ksuid2 True @@ -23,14 +23,18 @@ import secrets import time from datetime import datetime, timezone -from typing import Union, Optional +from typing import Optional __version__ = "1.0.0" __all__ = [ "KSUID", - "generate", "generate_lowercase", - "generate_token", "generate_token_lowercase", - "from_string", "from_base36", "from_bytes", + "generate", + "generate_lowercase", + "generate_token", + "generate_token_lowercase", + "from_string", + "from_base36", + "from_bytes", ] # KSUID epoch (May 13, 2014 16:53:20 UTC) @@ -38,7 +42,7 @@ # KSUID components TIMESTAMP_LENGTH = 4 # 4 bytes for timestamp -PAYLOAD_LENGTH = 16 # 16 bytes for random payload +PAYLOAD_LENGTH = 16 # 16 bytes for random payload TOTAL_LENGTH = TIMESTAMP_LENGTH + PAYLOAD_LENGTH # 20 bytes total # Base62 alphabet for encoding (mixed-case) @@ -63,55 +67,61 @@ class KSUID: KSUIDs are naturally sortable by creation time and collision-resistant. """ - __slots__ = ('_timestamp', '_payload', '_bytes') + __slots__ = ("_timestamp", "_payload", "_bytes") - def __init__(self, timestamp: Optional[int] = None, payload: Optional[bytes] = None): + def __init__( + self, timestamp: Optional[int] = None, payload: Optional[bytes] = None + ): """ Create a new KSUID. - + Args: timestamp: Unix timestamp (seconds). If None, uses current time. payload: 16-byte random payload. If None, generates random bytes. """ if timestamp is None: timestamp = int(time.time()) - + if payload is None: payload = os.urandom(PAYLOAD_LENGTH) elif len(payload) != PAYLOAD_LENGTH: raise ValueError(f"Payload must be exactly {PAYLOAD_LENGTH} bytes") - + # Convert timestamp to KSUID timestamp (relative to KSUID epoch) ksuid_timestamp = timestamp - EPOCH if ksuid_timestamp < 0: - raise ValueError("Timestamp cannot be before KSUID epoch (2014-05-13 16:53:20 UTC)") + raise ValueError( + "Timestamp cannot be before KSUID epoch (2014-05-13 16:53:20 UTC)" + ) if ksuid_timestamp >= 2**32: raise ValueError("Timestamp overflow: too far in the future") - + self._timestamp = ksuid_timestamp self._payload = payload - self._bytes = ksuid_timestamp.to_bytes(TIMESTAMP_LENGTH, 'big') + payload - + self._bytes = ksuid_timestamp.to_bytes(TIMESTAMP_LENGTH, "big") + payload + @classmethod - def from_string(cls, ksuid_str: str) -> 'KSUID': + def from_string(cls, ksuid_str: str) -> "KSUID": """ Create a KSUID from its string representation. - + Args: ksuid_str: Base62-encoded KSUID string - + Returns: KSUID instance """ if len(ksuid_str) != _BASE62_STRING_LENGTH: - raise ValueError(f"KSUID string must be exactly {_BASE62_STRING_LENGTH} characters") + raise ValueError( + f"KSUID string must be exactly {_BASE62_STRING_LENGTH} characters" + ) # Decode from base62 decoded_bytes = _base62_decode(ksuid_str) return cls.from_bytes(decoded_bytes) - + @classmethod - def from_base36(cls, ksuid_str: str) -> 'KSUID': + def from_base36(cls, ksuid_str: str) -> "KSUID": """ Create a KSUID from a lowercase base36 string representation. @@ -122,7 +132,10 @@ def from_base36(cls, ksuid_str: str) -> 'KSUID': KSUID instance """ if len(ksuid_str) != _BASE36_STRING_LENGTH: - raise ValueError(f"Base36 KSUID string must be exactly {_BASE36_STRING_LENGTH} characters") + raise ValueError( + f"Base36 KSUID string must be exactly " + f"{_BASE36_STRING_LENGTH} characters" + ) decoded_bytes = _base36_decode(ksuid_str) return cls.from_bytes(decoded_bytes) @@ -132,79 +145,79 @@ def to_base36(self) -> str: return _base36_encode(self._bytes) @classmethod - def from_bytes(cls, data: bytes) -> 'KSUID': + def from_bytes(cls, data: bytes) -> "KSUID": """ Create a KSUID from raw bytes. - + Args: data: 20-byte KSUID data - + Returns: KSUID instance """ if len(data) != TOTAL_LENGTH: raise ValueError(f"KSUID bytes must be exactly {TOTAL_LENGTH} bytes") - + timestamp_bytes = data[:TIMESTAMP_LENGTH] payload = data[TIMESTAMP_LENGTH:] - - ksuid_timestamp = int.from_bytes(timestamp_bytes, 'big') + + ksuid_timestamp = int.from_bytes(timestamp_bytes, "big") unix_timestamp = ksuid_timestamp + EPOCH - + return cls(unix_timestamp, payload) - + @property def timestamp(self) -> int: """Unix timestamp when this KSUID was created.""" return self._timestamp + EPOCH - + @property def datetime(self) -> datetime: """Datetime when this KSUID was created (UTC).""" return datetime.fromtimestamp(self.timestamp, tz=timezone.utc) - + @property def payload(self) -> bytes: """16-byte random payload.""" return self._payload - + @property def bytes(self) -> bytes: """Raw 20-byte KSUID data.""" return self._bytes - + def __str__(self) -> str: """Base62-encoded string representation.""" return _base62_encode(self._bytes) - + def __repr__(self) -> str: return f"KSUID('{str(self)}')" - + def __eq__(self, other) -> bool: if not isinstance(other, KSUID): return NotImplemented return self._bytes == other._bytes - + def __lt__(self, other) -> bool: if not isinstance(other, KSUID): return NotImplemented return self._bytes < other._bytes - + def __le__(self, other) -> bool: if not isinstance(other, KSUID): return NotImplemented return self._bytes <= other._bytes - + def __gt__(self, other) -> bool: if not isinstance(other, KSUID): return NotImplemented return self._bytes > other._bytes - + def __ge__(self, other) -> bool: if not isinstance(other, KSUID): return NotImplemented return self._bytes >= other._bytes - + def __hash__(self) -> int: return hash(self._bytes) @@ -215,16 +228,16 @@ def _base62_encode(data: bytes) -> str: return "" # Convert bytes to integer - num = int.from_bytes(data, 'big') + num = int.from_bytes(data, "big") result = [] while num > 0: num, remainder = divmod(num, BASE62_BASE) result.append(BASE62_ALPHABET[remainder]) - + # Pad to fixed width for KSUID result.reverse() - encoded = ''.join(result) + encoded = "".join(result) return encoded.zfill(_BASE62_STRING_LENGTH) @@ -250,7 +263,7 @@ def _base62_decode(s: str) -> bytes: raise ValueError("Base62 value exceeds maximum for KSUID") # Convert to bytes (20 bytes for KSUID) - return num.to_bytes(TOTAL_LENGTH, 'big') + return num.to_bytes(TOTAL_LENGTH, "big") # --- Base36 (lowercase) encoding --------------------------------------------------- @@ -263,7 +276,7 @@ def _base36_encode(data: bytes) -> str: if not data: return "" - num = int.from_bytes(data, 'big') + num = int.from_bytes(data, "big") result = [] while num > 0: @@ -271,7 +284,7 @@ def _base36_encode(data: bytes) -> str: result.append(BASE36_ALPHABET[remainder]) result.reverse() - encoded = ''.join(result) + encoded = "".join(result) return encoded.zfill(_BASE36_STRING_LENGTH) @@ -290,7 +303,7 @@ def _base36_decode(s: str) -> bytes: if num > _MAX_ENCODED: raise ValueError("Base36 value exceeds maximum for KSUID") - return num.to_bytes(TOTAL_LENGTH, 'big') + return num.to_bytes(TOTAL_LENGTH, "big") # Convenience functions @@ -349,4 +362,4 @@ def from_base36(ksuid_str: str) -> KSUID: def from_bytes(data: bytes) -> KSUID: """Create a KSUID from raw bytes.""" - return KSUID.from_bytes(data) \ No newline at end of file + return KSUID.from_bytes(data) diff --git a/prefixed_examples.py b/prefixed_examples.py index 071e3b7..b239e70 100644 --- a/prefixed_examples.py +++ b/prefixed_examples.py @@ -6,11 +6,7 @@ using KSUIDs for better developer experience and type safety. """ -import sys -import os -sys.path.insert(0, os.path.dirname(__file__)) - -from __init__ import KSUID, generate, generate_token, from_string +from ksuid import KSUID, generate, generate_token, from_string from typing import Dict, Optional, Tuple import re @@ -18,157 +14,155 @@ class PrefixedKSUID: """ A utility class for creating and managing prefixed KSUIDs like Stripe's API keys. - + Examples: user_2StGMtcWzRJ8qZqQjbJjGdTkVfv pi_2StGMtcWzRJ8qZqQjbJjGdTkVfv cus_2StGMtcWzRJ8qZqQjbJjGdTkVfv """ - + # Common prefix patterns used by various companies ENTITY_PREFIXES = { # User-related - 'user': 'user', # Users - 'admin': 'adm', # Administrators - 'guest': 'gst', # Guest users - + "user": "user", # Users + "admin": "adm", # Administrators + "guest": "gst", # Guest users # Payment-related (Stripe-style) - 'payment_intent': 'pi', - 'payment_method': 'pm', - 'customer': 'cus', - 'charge': 'ch', - 'refund': 're', - 'invoice': 'in', - 'subscription': 'sub', - 'product': 'prod', - 'price': 'price', - + "payment_intent": "pi", + "payment_method": "pm", + "customer": "cus", + "charge": "ch", + "refund": "re", + "invoice": "in", + "subscription": "sub", + "product": "prod", + "price": "price", # API-related - 'secret_key': 'sk', - 'public_key': 'pk', - 'api_key': 'ak', - 'token': 'tok', - 'session': 'sess', - + "secret_key": "sk", + "public_key": "pk", + "api_key": "ak", + "token": "tok", + "session": "sess", # Business entities - 'order': 'ord', - 'transaction': 'txn', - 'shipment': 'ship', - 'warehouse': 'wh', - 'inventory': 'inv', - + "order": "ord", + "transaction": "txn", + "shipment": "ship", + "warehouse": "wh", + "inventory": "inv", # Content-related - 'post': 'post', - 'comment': 'comm', - 'file': 'file', - 'upload': 'up', - 'download': 'dl', - + "post": "post", + "comment": "comm", + "file": "file", + "upload": "up", + "download": "dl", # System-related - 'log': 'log', - 'event': 'evt', - 'notification': 'notif', - 'webhook': 'whk', - 'job': 'job', - 'task': 'task' + "log": "log", + "event": "evt", + "notification": "notif", + "webhook": "whk", + "job": "job", + "task": "task", } - + @classmethod def create(cls, prefix: str) -> str: """ Create a prefixed KSUID. - + Args: prefix: The prefix to use (e.g., 'user', 'pi', 'cus') - + Returns: Prefixed KSUID string (e.g., 'user_2StGMtcWzRJ8qZqQjbJjGdTkVfv') """ if not prefix: raise ValueError("Prefix cannot be empty") - - # Validate prefix format (alphanumeric only, no underscores since _ is the delimiter) - if not re.match(r'^[a-zA-Z][a-zA-Z0-9]*$', prefix): - raise ValueError("Prefix must start with a letter and contain only alphanumeric characters") - + + # No underscores in prefix since _ is the delimiter + if not re.match(r"^[a-zA-Z][a-zA-Z0-9]*$", prefix): + raise ValueError( + "Prefix must start with a letter and " + "contain only alphanumeric characters" + ) + return f"{prefix}_{generate()}" - + @classmethod def parse(cls, prefixed_id: str) -> Tuple[str, KSUID]: """ Parse a prefixed KSUID into its components. - + Args: prefixed_id: The prefixed KSUID string - + Returns: Tuple of (prefix, KSUID object) - + Raises: ValueError: If the format is invalid """ - if not prefixed_id or '_' not in prefixed_id: + if not prefixed_id or "_" not in prefixed_id: raise ValueError("Invalid prefixed KSUID format") - - parts = prefixed_id.split('_', 1) + + parts = prefixed_id.split("_", 1) if len(parts) != 2: raise ValueError("Invalid prefixed KSUID format") - + prefix, ksuid_str = parts - + if not prefix: raise ValueError("Prefix cannot be empty") - + try: ksuid = from_string(ksuid_str) return prefix, ksuid except Exception as e: raise ValueError(f"Invalid KSUID part: {e}") - + @classmethod def validate(cls, prefixed_id: str, expected_prefix: Optional[str] = None) -> bool: """ Validate a prefixed KSUID. - + Args: prefixed_id: The prefixed KSUID to validate expected_prefix: Optional prefix to validate against - + Returns: True if valid, False otherwise """ try: prefix, ksuid = cls.parse(prefixed_id) - + if expected_prefix and prefix != expected_prefix: return False - + return True except ValueError: return False - + @classmethod def get_prefix(cls, prefixed_id: str) -> str: """ Extract just the prefix from a prefixed KSUID. - + Args: prefixed_id: The prefixed KSUID string - + Returns: The prefix part """ prefix, _ = cls.parse(prefixed_id) return prefix - + @classmethod def get_ksuid(cls, prefixed_id: str) -> KSUID: """ Extract just the KSUID from a prefixed KSUID. - + Args: prefixed_id: The prefixed KSUID string - + Returns: The KSUID object """ @@ -179,19 +173,23 @@ def get_ksuid(cls, prefixed_id: str) -> KSUID: # Convenience functions for common entity types def create_user_id() -> str: """Create a user ID: user_...""" - return PrefixedKSUID.create('user') + return PrefixedKSUID.create("user") + def create_payment_intent_id() -> str: """Create a payment intent ID: pi_...""" - return PrefixedKSUID.create('pi') + return PrefixedKSUID.create("pi") + def create_customer_id() -> str: """Create a customer ID: cus_...""" - return PrefixedKSUID.create('cus') + return PrefixedKSUID.create("cus") + def create_order_id() -> str: """Create an order ID: ord_...""" - return PrefixedKSUID.create('ord') + return PrefixedKSUID.create("ord") + def create_api_key() -> str: """Create a secure API key: ak_... @@ -201,6 +199,7 @@ def create_api_key() -> str: """ return f"ak_{generate_token()}" + def create_session_id() -> str: """Create a secure session token: sess_... @@ -213,19 +212,19 @@ def create_session_id() -> str: def demo_basic_usage(): """Demonstrate basic prefixed KSUID usage.""" print("=== Basic Prefixed KSUID Usage ===") - + # Create various types of IDs user_id = create_user_id() payment_id = create_payment_intent_id() customer_id = create_customer_id() order_id = create_order_id() - + print(f"User ID: {user_id}") print(f"Payment ID: {payment_id}") print(f"Customer ID: {customer_id}") print(f"Order ID: {order_id}") print() - + # Parse IDs print("Parsing IDs:") for prefixed_id in [user_id, payment_id, customer_id, order_id]: @@ -237,60 +236,64 @@ def demo_basic_usage(): def demo_validation(): """Demonstrate ID validation.""" print("=== ID Validation ===") - + user_id = create_user_id() payment_id = create_payment_intent_id() - + # Valid cases - print(f"Is '{user_id}' a valid user ID? {PrefixedKSUID.validate(user_id, 'user')}") - print(f"Is '{payment_id}' a valid payment ID? {PrefixedKSUID.validate(payment_id, 'pi')}") - + is_user = PrefixedKSUID.validate(user_id, "user") + print(f"Is '{user_id}' a valid user ID? {is_user}") + is_pay = PrefixedKSUID.validate(payment_id, "pi") + print(f"Is '{payment_id}' a valid payment ID? {is_pay}") + # Invalid cases print(f"Is '{user_id}' a valid payment ID? {PrefixedKSUID.validate(user_id, 'pi')}") print(f"Is 'invalid_id' valid? {PrefixedKSUID.validate('invalid_id')}") - print(f"Is 'user_invalid_ksuid' valid? {PrefixedKSUID.validate('user_invalid_ksuid')}") + print( + f"Is 'user_invalid_ksuid' valid? {PrefixedKSUID.validate('user_invalid_ksuid')}" + ) print() def demo_api_usage(): """Demonstrate API-style usage.""" print("=== API Usage Example ===") - + # Simulate API endpoints def create_user_endpoint(name: str, email: str) -> Dict: user_id = create_user_id() return { - 'id': user_id, - 'name': name, - 'email': email, - 'created_at': PrefixedKSUID.get_ksuid(user_id).datetime.isoformat() + "id": user_id, + "name": name, + "email": email, + "created_at": PrefixedKSUID.get_ksuid(user_id).datetime.isoformat(), } - + def create_payment_endpoint(user_id: str, amount: int) -> Dict: # Validate user ID - if not PrefixedKSUID.validate(user_id, 'user'): + if not PrefixedKSUID.validate(user_id, "user"): raise ValueError("Invalid user ID") - + payment_id = create_payment_intent_id() return { - 'id': payment_id, - 'user_id': user_id, - 'amount': amount, - 'status': 'pending', - 'created_at': PrefixedKSUID.get_ksuid(payment_id).datetime.isoformat() + "id": payment_id, + "user_id": user_id, + "amount": amount, + "status": "pending", + "created_at": PrefixedKSUID.get_ksuid(payment_id).datetime.isoformat(), } - + # Create user user = create_user_endpoint("John Doe", "john@example.com") print(f"Created user: {user}") - + # Create payment for user - payment = create_payment_endpoint(user['id'], 1000) + payment = create_payment_endpoint(user["id"], 1000) print(f"Created payment: {payment}") - + # Show chronological ordering - user_ksuid = PrefixedKSUID.get_ksuid(user['id']) - payment_ksuid = PrefixedKSUID.get_ksuid(payment['id']) + user_ksuid = PrefixedKSUID.get_ksuid(user["id"]) + payment_ksuid = PrefixedKSUID.get_ksuid(payment["id"]) print(f"User created before payment? {user_ksuid < payment_ksuid}") print() @@ -298,42 +301,40 @@ def create_payment_endpoint(user_id: str, amount: int) -> Dict: def demo_database_patterns(): """Demonstrate database usage patterns.""" print("=== Database Usage Patterns ===") - + # Simulate database records records = [] - + # Create mixed entity types for i in range(5): if i % 2 == 0: record_id = create_user_id() - record_type = 'user' + record_type = "user" else: record_id = create_order_id() - record_type = 'order' - - records.append({ - 'id': record_id, - 'type': record_type, - 'data': f'Sample {record_type} {i}' - }) - + record_type = "order" + + records.append( + {"id": record_id, "type": record_type, "data": f"Sample {record_type} {i}"} + ) + print("Created records:") for record in records: - ksuid = PrefixedKSUID.get_ksuid(record['id']) + ksuid = PrefixedKSUID.get_ksuid(record["id"]) print(f" {record['id']} ({record['type']}) - {ksuid.datetime}") - + # Sort by creation time (KSUID natural ordering) - sorted_records = sorted(records, key=lambda r: PrefixedKSUID.get_ksuid(r['id'])) - + sorted_records = sorted(records, key=lambda r: PrefixedKSUID.get_ksuid(r["id"])) + print("\nSorted by creation time:") for record in sorted_records: - ksuid = PrefixedKSUID.get_ksuid(record['id']) + ksuid = PrefixedKSUID.get_ksuid(record["id"]) print(f" {record['id']} ({record['type']}) - {ksuid.datetime}") - + # Filter by entity type - user_records = [r for r in records if PrefixedKSUID.get_prefix(r['id']) == 'user'] - order_records = [r for r in records if PrefixedKSUID.get_prefix(r['id']) == 'ord'] - + user_records = [r for r in records if PrefixedKSUID.get_prefix(r["id"]) == "user"] + order_records = [r for r in records if PrefixedKSUID.get_prefix(r["id"]) == "ord"] + print(f"\nUser records: {len(user_records)}") print(f"Order records: {len(order_records)}") print() @@ -342,7 +343,7 @@ def demo_database_patterns(): def demo_error_handling(): """Demonstrate error handling.""" print("=== Error Handling ===") - + test_cases = [ ("", "Empty string"), ("no_underscore", "No underscore"), @@ -350,9 +351,9 @@ def demo_error_handling(): ("user_", "Missing KSUID"), ("user_invalid_ksuid", "Invalid KSUID"), ("123_valid_ksuid", "Invalid prefix (starts with number)"), - ("user-invalid_valid_ksuid", "Invalid prefix (contains hyphen)") + ("user-invalid_valid_ksuid", "Invalid prefix (contains hyphen)"), ] - + for test_input, description in test_cases: try: prefix, ksuid = PrefixedKSUID.parse(test_input) @@ -367,13 +368,13 @@ def main(): print("Prefixed KSUID Examples - Stripe-Style Implementation") print("=" * 60) print() - + demo_basic_usage() demo_validation() demo_api_usage() demo_database_patterns() demo_error_handling() - + print("=== Summary ===") print("Prefixed KSUIDs provide:") print("✅ Type safety through prefixes") @@ -386,4 +387,4 @@ def main(): if __name__ == "__main__": - main() \ No newline at end of file + main() diff --git a/test_ksuid.py b/test_ksuid.py index 946f4db..fdae544 100644 --- a/test_ksuid.py +++ b/test_ksuid.py @@ -3,93 +3,101 @@ """ import time -import sys -import os -sys.path.insert(0, os.path.dirname(__file__)) import pytest from datetime import datetime, timezone -from __init__ import ( - KSUID, generate, generate_lowercase, generate_token, generate_token_lowercase, - from_string, from_base36, from_bytes, EPOCH, +from ksuid import ( + KSUID, + generate, + generate_lowercase, + generate_token, + generate_token_lowercase, + from_string, + from_base36, + from_bytes, + EPOCH, _BASE36_STRING_LENGTH, ) class TestKSUID: """Test cases for KSUID class.""" - + def test_generate_ksuid(self): """Test basic KSUID generation.""" ksuid = KSUID() - + # Check string representation length assert len(str(ksuid)) == 27 - + # Check bytes length assert len(ksuid.bytes) == 20 - + # Check payload length assert len(ksuid.payload) == 16 - + # Check timestamp is reasonable (within last minute) now = int(time.time()) assert abs(ksuid.timestamp - now) < 60 - + def test_ksuid_with_custom_timestamp(self): """Test KSUID creation with custom timestamp.""" custom_timestamp = 1609459200 # 2021-01-01 00:00:00 UTC ksuid = KSUID(timestamp=custom_timestamp) - + assert ksuid.timestamp == custom_timestamp - + # Check datetime conversion expected_dt = datetime.fromtimestamp(custom_timestamp, tz=timezone.utc) assert ksuid.datetime == expected_dt - + def test_ksuid_with_custom_payload(self): """Test KSUID creation with custom payload.""" - payload = b'\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10' + payload = b"\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10" ksuid = KSUID(payload=payload) - + assert ksuid.payload == payload - + def test_invalid_payload_length(self): """Test that invalid payload length raises error.""" with pytest.raises(ValueError, match="Payload must be exactly 16 bytes"): - KSUID(payload=b'\x01\x02\x03') # Too short - + KSUID(payload=b"\x01\x02\x03") # Too short + with pytest.raises(ValueError, match="Payload must be exactly 16 bytes"): - KSUID(payload=b'\x01' * 20) # Too long - + KSUID(payload=b"\x01" * 20) # Too long + def test_timestamp_before_epoch(self): """Test that timestamp before KSUID epoch raises error.""" with pytest.raises(ValueError, match="Timestamp cannot be before KSUID epoch"): KSUID(timestamp=EPOCH - 1) - + def test_timestamp_overflow(self): """Test that timestamp too far in future raises error.""" with pytest.raises(ValueError, match="Timestamp overflow"): KSUID(timestamp=EPOCH + 2**32) - + def test_from_string(self): """Test creating KSUID from string representation.""" ksuid1 = KSUID() ksuid_str = str(ksuid1) ksuid2 = KSUID.from_string(ksuid_str) - + assert ksuid1 == ksuid2 assert ksuid1.timestamp == ksuid2.timestamp assert ksuid1.payload == ksuid2.payload - + def test_from_string_invalid_length(self): """Test that invalid string length raises error.""" - with pytest.raises(ValueError, match="KSUID string must be exactly 27 characters"): + with pytest.raises( + ValueError, match="KSUID string must be exactly 27 characters" + ): KSUID.from_string("too_short") - - with pytest.raises(ValueError, match="KSUID string must be exactly 27 characters"): + + with pytest.raises( + ValueError, match="KSUID string must be exactly 27 characters" + ): KSUID.from_string("a" * 30) # Too long - + def test_from_string_invalid_characters(self): """Test that invalid base62 characters raise error.""" with pytest.raises(ValueError, match="Invalid base62 character"): @@ -99,48 +107,48 @@ def test_from_string_overflow(self): """Test that a base62 string exceeding 20-byte max raises ValueError.""" with pytest.raises(ValueError, match="Base62 value exceeds maximum"): KSUID.from_string("z" * 27) # Exceeds 2^160 - 1 - + def test_from_bytes(self): """Test creating KSUID from bytes.""" ksuid1 = KSUID() ksuid_bytes = ksuid1.bytes ksuid2 = KSUID.from_bytes(ksuid_bytes) - + assert ksuid1 == ksuid2 assert ksuid1.timestamp == ksuid2.timestamp assert ksuid1.payload == ksuid2.payload - + def test_from_bytes_invalid_length(self): """Test that invalid bytes length raises error.""" with pytest.raises(ValueError, match="KSUID bytes must be exactly 20 bytes"): - KSUID.from_bytes(b'\x01\x02\x03') # Too short - + KSUID.from_bytes(b"\x01\x02\x03") # Too short + with pytest.raises(ValueError, match="KSUID bytes must be exactly 20 bytes"): - KSUID.from_bytes(b'\x01' * 25) # Too long - + KSUID.from_bytes(b"\x01" * 25) # Too long + def test_sortability(self): """Test that KSUIDs are sortable by creation time.""" # Create KSUIDs with different timestamps ksuid1 = KSUID(timestamp=1609459200) # 2021-01-01 ksuid2 = KSUID(timestamp=1609459201) # 2021-01-01 + 1 second ksuid3 = KSUID(timestamp=1609459202) # 2021-01-01 + 2 seconds - + # Test all comparison operators assert ksuid1 < ksuid2 < ksuid3 assert ksuid1 <= ksuid2 <= ksuid3 assert ksuid3 > ksuid2 > ksuid1 assert ksuid3 >= ksuid2 >= ksuid1 - + # Test sorting ksuids = [ksuid3, ksuid1, ksuid2] sorted_ksuids = sorted(ksuids) assert sorted_ksuids == [ksuid1, ksuid2, ksuid3] - + def test_equality(self): """Test KSUID equality.""" # Same timestamp and payload should be equal timestamp = 1609459200 - payload = b'\x01' * 16 + payload = b"\x01" * 16 ksuid1 = KSUID(timestamp=timestamp, payload=payload) ksuid2 = KSUID(timestamp=timestamp, payload=payload) @@ -148,7 +156,7 @@ def test_equality(self): assert hash(ksuid1) == hash(ksuid2) # Different payload should not be equal - ksuid3 = KSUID(timestamp=timestamp, payload=b'\x02' * 16) + ksuid3 = KSUID(timestamp=timestamp, payload=b"\x02" * 16) assert ksuid1 != ksuid3 def test_equality_with_non_ksuid(self): @@ -157,102 +165,104 @@ def test_equality_with_non_ksuid(self): assert ksuid.__eq__("not a ksuid") is NotImplemented assert ksuid.__eq__(42) is NotImplemented assert ksuid.__eq__(None) is NotImplemented - + def test_string_representation(self): """Test string and repr methods.""" ksuid = KSUID() ksuid_str = str(ksuid) - + # String should be 27 characters assert len(ksuid_str) == 27 - + # Repr should contain the string assert ksuid_str in repr(ksuid) assert "KSUID" in repr(ksuid) - + def test_round_trip_conversion(self): """Test that string/bytes conversions are reversible.""" ksuid1 = KSUID() - + # String round trip ksuid_str = str(ksuid1) ksuid2 = KSUID.from_string(ksuid_str) assert ksuid1 == ksuid2 - + # Bytes round trip ksuid_bytes = ksuid1.bytes ksuid3 = KSUID.from_bytes(ksuid_bytes) assert ksuid1 == ksuid3 - + def test_datetime_property(self): """Test datetime property conversion.""" timestamp = 1609459200 # 2021-01-01 00:00:00 UTC ksuid = KSUID(timestamp=timestamp) - + expected_dt = datetime(2021, 1, 1, 0, 0, 0, tzinfo=timezone.utc) assert ksuid.datetime == expected_dt class TestConvenienceFunctions: """Test convenience functions.""" - + def test_generate(self): """Test generate() function.""" ksuid = generate() assert isinstance(ksuid, KSUID) assert len(str(ksuid)) == 27 - + def test_from_string_function(self): """Test from_string() function.""" ksuid1 = generate() ksuid_str = str(ksuid1) ksuid2 = from_string(ksuid_str) - + assert ksuid1 == ksuid2 - + def test_from_bytes_function(self): """Test from_bytes() function.""" ksuid1 = generate() ksuid_bytes = ksuid1.bytes ksuid2 = from_bytes(ksuid_bytes) - + assert ksuid1 == ksuid2 class TestKSUIDProperties: """Test KSUID properties and edge cases.""" - + def test_multiple_ksuids_different(self): """Test that multiple KSUIDs generated quickly are different.""" ksuids = [generate() for _ in range(100)] - + # All should be unique assert len(set(ksuids)) == 100 - + # All should be sortable (no exceptions) sorted_ksuids = sorted(ksuids) assert len(sorted_ksuids) == 100 - + def test_ksuid_ordering_with_same_timestamp(self): """Test KSUID ordering when timestamps are the same.""" timestamp = 1609459200 - + # Create KSUIDs with same timestamp but different payloads - ksuid1 = KSUID(timestamp=timestamp, payload=b'\x00' * 16) - ksuid2 = KSUID(timestamp=timestamp, payload=b'\x01' * 16) - + ksuid1 = KSUID(timestamp=timestamp, payload=b"\x00" * 16) + ksuid2 = KSUID(timestamp=timestamp, payload=b"\x01" * 16) + # They should still be comparable (by payload) assert ksuid1 < ksuid2 - + def test_base62_encoding_properties(self): """Test properties of base62 encoding.""" ksuid = generate() ksuid_str = str(ksuid) - + # Should only contain base62 characters - valid_chars = set("0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz") + valid_chars = set( + "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + ) assert all(c in valid_chars for c in ksuid_str) - + # Should be exactly 27 characters assert len(ksuid_str) == 27 @@ -284,15 +294,18 @@ def test_token_differs_from_ksuid_structure(self): range. We generate 10 tokens and verify none of them decode to a timestamp within 1 year of now (probability ≈ (2/2^32)^10). """ - from __init__ import _base62_decode + from ksuid import _base62_decode + now = int(time.time()) one_year = 365 * 86400 for _ in range(10): raw = _base62_decode(generate_token()) - ts_value = int.from_bytes(raw[:4], 'big') + EPOCH + ts_value = int.from_bytes(raw[:4], "big") + EPOCH if abs(ts_value - now) > one_year: return # At least one clearly non-timestamp token — pass - pytest.fail("All 10 tokens decoded to timestamps near 'now' — extremely unlikely") + pytest.fail( + "All 10 tokens decoded to timestamps near 'now' — extremely unlikely" + ) class TestSlots: @@ -301,7 +314,7 @@ class TestSlots: def test_no_instance_dict(self): """KSUID instances should not have a __dict__.""" ksuid = KSUID() - assert not hasattr(ksuid, '__dict__') + assert not hasattr(ksuid, "__dict__") def test_cannot_set_arbitrary_attribute(self): """Setting an undefined attribute should raise AttributeError.""" @@ -316,6 +329,7 @@ class TestThreadSafety: def test_concurrent_generate(self): """KSUIDs generated across threads must all be unique.""" from concurrent.futures import ThreadPoolExecutor + count_per_thread = 500 num_threads = 4 @@ -332,6 +346,7 @@ def gen_batch(_): def test_concurrent_generate_token(self): """Tokens generated across threads must all be unique.""" from concurrent.futures import ThreadPoolExecutor + count_per_thread = 500 num_threads = 4 @@ -403,7 +418,7 @@ def test_from_base36_overflow(self): def test_lowercase_sortability(self): """Base36 strings of KSUIDs with increasing timestamps must sort.""" ts1, ts2, ts3 = 1609459200, 1609459201, 1609459202 - payload = b'\x00' * 16 + payload = b"\x00" * 16 s1 = KSUID(timestamp=ts1, payload=payload).to_base36() s2 = KSUID(timestamp=ts2, payload=payload).to_base36() s3 = KSUID(timestamp=ts3, payload=payload).to_base36() @@ -431,15 +446,15 @@ def test_generate_token_lowercase_uniqueness(self): def test_zero_value_base36_round_trip(self): """All-zero KSUID must encode to 31 '0' chars in base36.""" - ksuid = KSUID(timestamp=EPOCH, payload=b'\x00' * 16) + ksuid = KSUID(timestamp=EPOCH, payload=b"\x00" * 16) s = ksuid.to_base36() assert len(s) == _BASE36_STRING_LENGTH - assert s == '0' * _BASE36_STRING_LENGTH + assert s == "0" * _BASE36_STRING_LENGTH assert KSUID.from_base36(s) == ksuid def test_max_value_base36_round_trip(self): """Max-timestamp, max-payload KSUID must round-trip via base36.""" - ksuid = KSUID(timestamp=EPOCH + 2**32 - 1, payload=b'\xff' * 16) + ksuid = KSUID(timestamp=EPOCH + 2**32 - 1, payload=b"\xff" * 16) s = ksuid.to_base36() assert len(s) == _BASE36_STRING_LENGTH assert KSUID.from_base36(s) == ksuid @@ -450,42 +465,39 @@ class TestEdgeCases: def test_zero_value_round_trip(self): """All-zero KSUID must encode to 27 chars and round-trip correctly.""" - ksuid = KSUID(timestamp=EPOCH, payload=b'\x00' * 16) + ksuid = KSUID(timestamp=EPOCH, payload=b"\x00" * 16) s = str(ksuid) assert len(s) == 27 - assert s == '0' * 27 + assert s == "0" * 27 assert KSUID.from_string(s) == ksuid def test_max_timestamp_round_trip(self): """KSUID at the maximum timestamp must round-trip correctly.""" max_ts = EPOCH + 2**32 - 1 - ksuid = KSUID(timestamp=max_ts, payload=b'\xff' * 16) + ksuid = KSUID(timestamp=max_ts, payload=b"\xff" * 16) s = str(ksuid) assert len(s) == 27 assert KSUID.from_string(s) == ksuid if __name__ == "__main__": - # Run tests if script is executed directly - import sys - # Basic smoke test print("Running basic KSUID tests...") - + # Test generation ksuid1 = generate() print(f"Generated KSUID: {ksuid1}") print(f"Timestamp: {ksuid1.datetime}") - + # Test round-trip ksuid2 = from_string(str(ksuid1)) assert ksuid1 == ksuid2 print("Round-trip test passed!") - + # Test sortability (use explicit timestamps to avoid flaky 1ms sleep) - earlier = KSUID(timestamp=1609459200, payload=b'\x00' * 16) - later = KSUID(timestamp=1609459201, payload=b'\x00' * 16) + earlier = KSUID(timestamp=1609459200, payload=b"\x00" * 16) + later = KSUID(timestamp=1609459201, payload=b"\x00" * 16) assert earlier < later print("Sortability test passed!") - - print("\nAll basic tests passed! Run with pytest for comprehensive testing.") \ No newline at end of file + + print("\nAll basic tests passed! Run with pytest for comprehensive testing.") From 297848ed76327e3ddea0f36d3f9b9b23b3631c62 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 6 Feb 2026 18:54:23 +0000 Subject: [PATCH 3/3] Replace Unicode emojis with ASCII in print statements Windows cp1252 codepage cannot encode Unicode checkmark (U+2705) and cross mark (U+274C), causing UnicodeEncodeError on Windows CI runners. Replace all emoji in print() calls with ASCII equivalents ([+], [ok], [err]). https://claude.ai/code/session_01PVPVUNWhpxVa3xbDDBnwp2 --- benchmark.py | 12 ++++++------ example.py | 12 ++++++------ prefixed_examples.py | 18 +++++++++--------- 3 files changed, 21 insertions(+), 21 deletions(-) diff --git a/benchmark.py b/benchmark.py index 64c9f08..ee10dc1 100644 --- a/benchmark.py +++ b/benchmark.py @@ -178,12 +178,12 @@ def main(): print("=== Performance Summary ===") print("KSUID operations are highly optimized:") - print("✅ Generation: ~300k+ KSUIDs/second") - print("✅ String parsing: ~500k+ parses/second") - print("✅ Bytes parsing: ~1M+ parses/second") - print("✅ Comparison: ~10M+ comparisons/second") - print("✅ Sorting: ~500k+ items/second") - print("✅ Memory efficient: ~100 bytes per KSUID including overhead") + print("[+] Generation: ~300k+ KSUIDs/second") + print("[+] String parsing: ~500k+ parses/second") + print("[+] Bytes parsing: ~1M+ parses/second") + print("[+] Comparison: ~10M+ comparisons/second") + print("[+] Sorting: ~500k+ items/second") + print("[+] Memory efficient: ~100 bytes per KSUID including overhead") if __name__ == "__main__": diff --git a/example.py b/example.py index 9fb0546..5474e0e 100644 --- a/example.py +++ b/example.py @@ -197,12 +197,12 @@ def main(): print("=== Summary ===") print("KSUIDs provide:") - print("✅ Sortable unique identifiers") - print("✅ Compact 27-character representation") - print("✅ URL-safe base62 encoding") - print("✅ Embedded timestamp for debugging") - print("✅ High performance and collision resistance") - print("✅ Perfect for distributed systems and databases") + print("[+] Sortable unique identifiers") + print("[+] Compact 27-character representation") + print("[+] URL-safe base62 encoding") + print("[+] Embedded timestamp for debugging") + print("[+] High performance and collision resistance") + print("[+] Perfect for distributed systems and databases") if __name__ == "__main__": diff --git a/prefixed_examples.py b/prefixed_examples.py index b239e70..f251276 100644 --- a/prefixed_examples.py +++ b/prefixed_examples.py @@ -357,9 +357,9 @@ def demo_error_handling(): for test_input, description in test_cases: try: prefix, ksuid = PrefixedKSUID.parse(test_input) - print(f"✅ {description}: Parsed successfully") + print(f"[ok] {description}: Parsed successfully") except ValueError as e: - print(f"❌ {description}: {e}") + print(f"[err] {description}: {e}") print() @@ -377,13 +377,13 @@ def main(): print("=== Summary ===") print("Prefixed KSUIDs provide:") - print("✅ Type safety through prefixes") - print("✅ Chronological ordering") - print("✅ URL-safe identifiers") - print("✅ Easy validation and parsing") - print("✅ Database-friendly design") - print("✅ Developer-friendly APIs") - print("✅ Industry-standard patterns") + print("[+] Type safety through prefixes") + print("[+] Chronological ordering") + print("[+] URL-safe identifiers") + print("[+] Easy validation and parsing") + print("[+] Database-friendly design") + print("[+] Developer-friendly APIs") + print("[+] Industry-standard patterns") if __name__ == "__main__":