-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathexample.py
More file actions
209 lines (160 loc) · 6 KB
/
example.py
File metadata and controls
209 lines (160 loc) · 6 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
#!/usr/bin/env python3
"""
Example usage of the KSUID library.
This script demonstrates various features of the KSUID library including
generation, parsing, sorting, and practical usage patterns.
"""
import time
from datetime import datetime, timezone
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}")
print(f"Length: {len(str(ksuid1))} characters")
print(f"Timestamp: {ksuid1.datetime}")
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)
print(f"Recreated from string: {ksuid2}")
print(f"Are they equal? {ksuid1 == ksuid2}")
print()
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} "
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}")
print()
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}")
print()
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:,}")
print(f"Collision rate: {(count - len(unique_ksuids))/count*100:.6f}%")
print()
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
}
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']}"
)
# 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"])
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]
print(f"Found {len(recent_users)} recent users")
print()
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}")
print()
def main():
"""Run all demonstrations."""
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")
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__":
main()