mirrors/cpython
0
0 Fork 0
mirror of https://github.com/python/cpython.git synced 2026-10-28 10:25:05 +08:00
Code Activity

gh-139122: Reimplement base UUID type, uuid4(), and uuid7() in C (#139123)

Browse Source 
The C implementation considerably boosts the performance of the key UUID operations:

------------------------------------
Operation                    Speedup
------------------------------------
uuid4() generation            15.01x
uuid7() generation            29.64x
UUID from string               6.76x
UUID from bytes                5.16x
str(uuid) conversion           6.66x
------------------------------------

Summary of changes:

* The UUID type is reimplemented in C in its entirety.

* The pure-Python is kept around and is used of the C implementation
  isn't available for some reason.

* Both implementations are tested extensively; additional tests are
  added to ensure that the C implementation of the type follows the pure
  Python implementation fully.

* The Python implementation stores UUID values as int objects. The C
  implementation stores them as `uint8_t[16]` array.

* The C implementation has faster hash() implementation but also caches
  the computed hash value to speedup cases when UUIDs are used as
  set/dict keys.

* The C implementation has a freelist to make new UUID object
  instantiation as fast as possible.

* uuid4() and uuid7() are now implmented in C. The most performance
  boost (10x) comes from overfetching entropy to decrease the number of
  _PyOS_URandom() calls. On its own it's a safe optimization with the
  edge case that Unix fork needs to be explicitly handled. We do that by
  comparing the current PID to the PID of when the random buffer was
  populated.

* Portions of code are coming from my implementation of faster UUID
  in gel-python [1]. I did use AI during the development, but basically
  had to rewrite the code it generated to be more idiomatic and
  efficient.

* The benchmark can be found here [2].

* This PR makes Python UUID operations as fast as they are in NodeJS and
  Bun runtimes.

[1] https://github.com/MagicStack/py-pgproto/blob/b8109fb311a59f30f9947567a13508da9a776564/uuid.pyx

[2] https://gist.github.com/1st1/f03e816f34a61e4d46c78ff98baf4818
This commit is contained in:
Yury Selivanov authored and GitHub committed 2026-10-21 14:17:07 -07:00
1 parent 239808c189
commit cf690004e3
13 files changed
+2822 -43
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Doc/library/uuid.rst
+5
View File
@@ -69,6 +69,11 @@ which relays any information about the UUID's safety, using this enumeration:
variant and version number set according to :rfc:`9562`, overriding bits in the
given *hex*, *bytes*, *bytes_le*, *fields*, or *int*.
.. versionchanged:: next
Passing an object other than :class:`bytes` as *bytes* or *bytes_le*
now consistently raises :exc:`TypeError`. Previously, some invalid inputs
raised :exc:`AssertionError` or :exc:`ValueError`.
Comparison of UUID objects are made by way of comparing their
:attr:`UUID.int` attributes. Comparison with a non-UUID object
raises a :exc:`TypeError`.
Doc/whatsnew/3.16.rst
+40
View File
@@ -852,6 +852,18 @@ unicodedata
* The Unicode database has been updated to Unicode 18.0.0.
uuid
----
* :class:`uuid.UUID`, :func:`uuid.uuid4`, and :func:`uuid.uuid7` now have
C implementations, with a pure Python fallback.
In microbenchmarks, UUIDv4 and UUIDv7 generation are approximately 15 and
30 times faster, respectively, than the pure Python implementations.
UUID construction and string conversion are also faster; see
:ref:`whatsnew316-uuid-optimizations` for the benchmark results.
(Contributed by Yury Selivanov in :gh:`139123`.)
xml
---
@@ -960,6 +972,28 @@ re
:gh:`152056`.)
.. _whatsnew316-uuid-optimizations:
uuid
----
* C implementations accelerate UUID generation, construction, and string
conversion. Microbenchmarks comparing the C and pure Python implementations
measured the following speedups:
================================ =======
Operation Speedup
================================ =======
:func:`uuid.uuid4` generation 15.01x
:func:`uuid.uuid7` generation 29.64x
:class:`uuid.UUID` from a string 6.76x
:class:`uuid.UUID` from bytes 5.16x
:func:`str` conversion of a UUID 6.66x
================================ =======
(Contributed by Yury Selivanov in :gh:`139123`.)
Removed
========
@@ -1158,6 +1192,12 @@ that may require changes to your code.
:exc:`TypeError`.
(Contributed by Serhiy Storchaka in :gh:`152587`.)
* :class:`uuid.UUID` now consistently raises :exc:`TypeError` when the
*bytes* or *bytes_le* argument is not a :class:`bytes` object.
Previously, some invalid inputs raised :exc:`AssertionError` or
:exc:`ValueError`.
(Contributed by Yury Selivanov in :gh:`139123`.)
* :mod:`xml.dom.minidom` now raises :exc:`~xml.dom.InvalidCharacterErr`
for a name which is not a valid XML name,
:exc:`~xml.dom.WrongDocumentErr`
Include/internal/pycore_global_objects_fini_generated.h
Generated
+6
View File
@@ -337,6 +337,7 @@ _PyStaticObjects_CheckAll(PyInterpreterState *interp) {
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(byte_size), "byte_size", 9);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(byteorder), "byteorder", 9);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(bytes), "bytes", 5);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(bytes_le), "bytes_le", 8);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(bytes_per_sep), "bytes_per_sep", 13);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(c_call), "c_call", 6);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(c_exception), "c_exception", 11);
@@ -531,6 +532,7 @@ _PyStaticObjects_CheckAll(PyInterpreterState *interp) {
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(hash_name), "hash_name", 9);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(header), "header", 6);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(headers), "headers", 7);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(hex), "hex", 3);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(hi), "hi", 2);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(hook), "hook", 4);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(host), "host", 4);
@@ -564,6 +566,7 @@ _PyStaticObjects_CheckAll(PyInterpreterState *interp) {
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(insert_comments), "insert_comments", 15);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(insert_pis), "insert_pis", 10);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(instructions), "instructions", 12);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(int), "int", 3);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(intern), "intern", 6);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(intersection), "intersection", 12);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(interval), "interval", 8);
@@ -571,6 +574,7 @@ _PyStaticObjects_CheckAll(PyInterpreterState *interp) {
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(is_compress), "is_compress", 11);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(is_raw), "is_raw", 6);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(is_running), "is_running", 10);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(is_safe), "is_safe", 7);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(is_struct), "is_struct", 9);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(isatty), "isatty", 6);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(isinstance), "isinstance", 10);
@@ -744,6 +748,7 @@ _PyStaticObjects_CheckAll(PyInterpreterState *interp) {
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(query), "query", 5);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(queuetype), "queuetype", 9);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(quotetabs), "quotetabs", 9);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(random_func), "random_func", 11);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(raw), "raw", 3);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(read), "read", 4);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(read1), "read1", 5);
@@ -858,6 +863,7 @@ _PyStaticObjects_CheckAll(PyInterpreterState *interp) {
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(threading), "threading", 9);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(throw), "throw", 5);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(time), "time", 4);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(time_func), "time_func", 9);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(timeout), "timeout", 7);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(timer), "timer", 5);
_PyStaticObject_CheckUnicodeSingleton((PyObject *)&_Py_ID(times), "times", 5);
Include/internal/pycore_global_strings.h
+6
View File
@@ -349,6 +349,7 @@ struct _Py_global_strings {
STRUCT_FOR_ID(byte_size)
STRUCT_FOR_ID(byteorder)
STRUCT_FOR_ID(bytes)
STRUCT_FOR_ID(bytes_le)
STRUCT_FOR_ID(bytes_per_sep)
STRUCT_FOR_ID(c_call)
STRUCT_FOR_ID(c_exception)
@@ -543,6 +544,7 @@ struct _Py_global_strings {
STRUCT_FOR_ID(hash_name)
STRUCT_FOR_ID(header)
STRUCT_FOR_ID(headers)
STRUCT_FOR_ID(hex)
STRUCT_FOR_ID(hi)
STRUCT_FOR_ID(hook)
STRUCT_FOR_ID(host)
@@ -576,6 +578,7 @@ struct _Py_global_strings {
STRUCT_FOR_ID(insert_comments)
STRUCT_FOR_ID(insert_pis)
STRUCT_FOR_ID(instructions)
STRUCT_FOR_ID(int)
STRUCT_FOR_ID(intern)
STRUCT_FOR_ID(intersection)
STRUCT_FOR_ID(interval)
@@ -583,6 +586,7 @@ struct _Py_global_strings {
STRUCT_FOR_ID(is_compress)
STRUCT_FOR_ID(is_raw)
STRUCT_FOR_ID(is_running)
STRUCT_FOR_ID(is_safe)
STRUCT_FOR_ID(is_struct)
STRUCT_FOR_ID(isatty)
STRUCT_FOR_ID(isinstance)
@@ -756,6 +760,7 @@ struct _Py_global_strings {
STRUCT_FOR_ID(query)
STRUCT_FOR_ID(queuetype)
STRUCT_FOR_ID(quotetabs)
STRUCT_FOR_ID(random_func)
STRUCT_FOR_ID(raw)
STRUCT_FOR_ID(read)
STRUCT_FOR_ID(read1)
@@ -870,6 +875,7 @@ struct _Py_global_strings {
STRUCT_FOR_ID(threading)
STRUCT_FOR_ID(throw)
STRUCT_FOR_ID(time)
STRUCT_FOR_ID(time_func)
STRUCT_FOR_ID(timeout)
STRUCT_FOR_ID(timer)
STRUCT_FOR_ID(times)
Include/internal/pycore_pylifecycle.h
+2 -2
View File
@@ -96,8 +96,8 @@ extern const char* _Py_gitversion(void);
// Export for '_asyncio' shared extension
PyAPI_FUNC(int) _Py_IsInterpreterFinalizing(PyInterpreterState *interp);
/* Random */
extern int _PyOS_URandom(void *buffer, Py_ssize_t size);
// Export for '_uuid' shared extension
PyAPI_FUNC(int) _PyOS_URandom(void *buffer, Py_ssize_t size);
// Export for '_random' shared extension
PyAPI_FUNC(int) _PyOS_URandomNonblock(void *buffer, Py_ssize_t size);
Include/internal/pycore_runtime_init_generated.h
Generated
+6
View File
@@ -1624,6 +1624,7 @@ extern "C" {
INIT_ID(byte_size), \
INIT_ID(byteorder), \
INIT_ID(bytes), \
INIT_ID(bytes_le), \
INIT_ID(bytes_per_sep), \
INIT_ID(c_call), \
INIT_ID(c_exception), \
@@ -1818,6 +1819,7 @@ extern "C" {
INIT_ID(hash_name), \
INIT_ID(header), \
INIT_ID(headers), \
INIT_ID(hex), \
INIT_ID(hi), \
INIT_ID(hook), \
INIT_ID(host), \
@@ -1851,6 +1853,7 @@ extern "C" {
INIT_ID(insert_comments), \
INIT_ID(insert_pis), \
INIT_ID(instructions), \
INIT_ID(int), \
INIT_ID(intern), \
INIT_ID(intersection), \
INIT_ID(interval), \
@@ -1858,6 +1861,7 @@ extern "C" {
INIT_ID(is_compress), \
INIT_ID(is_raw), \
INIT_ID(is_running), \
INIT_ID(is_safe), \
INIT_ID(is_struct), \
INIT_ID(isatty), \
INIT_ID(isinstance), \
@@ -2031,6 +2035,7 @@ extern "C" {
INIT_ID(query), \
INIT_ID(queuetype), \
INIT_ID(quotetabs), \
INIT_ID(random_func), \
INIT_ID(raw), \
INIT_ID(read), \
INIT_ID(read1), \
@@ -2145,6 +2150,7 @@ extern "C" {
INIT_ID(threading), \
INIT_ID(throw), \
INIT_ID(time), \
INIT_ID(time_func), \
INIT_ID(timeout), \
INIT_ID(timer), \
INIT_ID(times), \
Include/internal/pycore_unicodeobject_generated.h
Generated
+6
View File
@@ -301,6 +301,7 @@ _PyUnicode_InitStaticStrings(PyInterpreterState *interp) {
offsetof(struct _Py_global_strings, identifiers._py_byte_size._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_byteorder._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_bytes._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_bytes_le._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_bytes_per_sep._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_c_call._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_c_exception._ascii.ob_base),
@@ -495,6 +496,7 @@ _PyUnicode_InitStaticStrings(PyInterpreterState *interp) {
offsetof(struct _Py_global_strings, identifiers._py_hash_name._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_header._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_headers._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_hex._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_hi._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_hook._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_host._ascii.ob_base),
@@ -528,6 +530,7 @@ _PyUnicode_InitStaticStrings(PyInterpreterState *interp) {
offsetof(struct _Py_global_strings, identifiers._py_insert_comments._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_insert_pis._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_instructions._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_int._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_intern._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_intersection._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_interval._ascii.ob_base),
@@ -535,6 +538,7 @@ _PyUnicode_InitStaticStrings(PyInterpreterState *interp) {
offsetof(struct _Py_global_strings, identifiers._py_is_compress._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_is_raw._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_is_running._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_is_safe._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_is_struct._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_isatty._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_isinstance._ascii.ob_base),
@@ -708,6 +712,7 @@ _PyUnicode_InitStaticStrings(PyInterpreterState *interp) {
offsetof(struct _Py_global_strings, identifiers._py_query._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_queuetype._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_quotetabs._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_random_func._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_raw._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_read._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_read1._ascii.ob_base),
@@ -822,6 +827,7 @@ _PyUnicode_InitStaticStrings(PyInterpreterState *interp) {
offsetof(struct _Py_global_strings, identifiers._py_threading._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_throw._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_time._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_time_func._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_timeout._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_timer._ascii.ob_base),
offsetof(struct _Py_global_strings, identifiers._py_times._ascii.ob_base),
Lib/test/test_uuid.py
+621 -11
View File
@@ -7,13 +7,19 @@ import os
import pickle
import random
import sys
import textwrap
import unittest
import weakref
from collections import UserDict
from itertools import product
from types import MappingProxyType
from unittest import mock
from test import support
from test.support import force_not_colorized_test_class, import_helper, warnings_helper
from test.support import (
force_not_colorized_test_class, import_helper, threading_helper,
warnings_helper,
)
from test.support.script_helper import assert_python_ok
py_uuid = import_helper.import_fresh_module('uuid', blocked=['_uuid'])
@@ -35,6 +41,7 @@ def mock_get_command_stdout(data):
class BaseTestUUID:
uuid = None
is_c_uuid = False
def test_nil_uuid(self):
nil_uuid = self.uuid.NIL
@@ -270,6 +277,40 @@ class BaseTestUUID:
resorted.sort()
equal(ascending, resorted)
def test_UUID_prefixes(self):
# Python removes both URN components regardless of their order. Check
# that both implementations accept them with each UUID string format.
value = '12345678-1234-5678-1234-567812345678'
expected = self.uuid.UUID(value)
for prefix in ('', 'urn:', 'uuid:', 'urn:uuid:', 'uuid:urn:'):
for text in (value, value.replace('-', ''), '{' + value + '}'):
with self.subTest(prefix=prefix, text=text):
self.assertEqual(self.uuid.UUID(prefix + text), expected)
def test_UUID_strip_braces(self):
# str.strip('{}') removes runs of either brace at both ends, without
# requiring matching pairs. Keep the C parser's behavior identical.
value = '12345678-1234-5678-1234-567812345678'
expected = self.uuid.UUID(value)
for leading in ('', '{', '}', '{{{', '}}}', '{}}{'):
for trailing in ('', '{', '}', '{{{', '}}}', '{}}{'):
with self.subTest(leading=leading, trailing=trailing):
self.assertEqual(
self.uuid.UUID(leading + value + trailing), expected)
# Braces within the UUID and inputs containing only braces are invalid.
for text in ('{{}}', '{}{}', value[:8] + '{}' + value[8:]):
with self.subTest(text=text):
with self.assertRaises(ValueError):
self.uuid.UUID(text)
def test_repr_subclass_str(self):
# repr() must call the subclass's __str__, as the Python version does.
class CustomUUID(self.uuid.UUID):
def __str__(self):
return 'custom uuid'
self.assertEqual(repr(CustomUUID(int=1)), "CustomUUID('custom uuid')")
def test_exceptions(self):
badvalue = lambda f: self.assertRaises(ValueError, f)
badtype = lambda f: self.assertRaises(TypeError, f)
@@ -282,14 +323,16 @@ class BaseTestUUID:
badvalue(lambda: self.uuid.UUID('123456781234567812345678z2345678'))
# Badly formed bytes.
badvalue(lambda: self.uuid.UUID(bytes='abc'))
badvalue(lambda: self.uuid.UUID(bytes='\0'*15))
badvalue(lambda: self.uuid.UUID(bytes='\0'*17))
badtype(lambda: self.uuid.UUID(bytes='abc'))
badvalue(lambda: self.uuid.UUID(bytes=b'abc'))
badvalue(lambda: self.uuid.UUID(bytes=b'\0'*15))
badvalue(lambda: self.uuid.UUID(bytes=b'\0'*17))
# Badly formed bytes_le.
badvalue(lambda: self.uuid.UUID(bytes_le='abc'))
badvalue(lambda: self.uuid.UUID(bytes_le='\0'*15))
badvalue(lambda: self.uuid.UUID(bytes_le='\0'*17))
badtype(lambda: self.uuid.UUID(bytes_le='abc'))
badvalue(lambda: self.uuid.UUID(bytes_le=b'abc'))
badvalue(lambda: self.uuid.UUID(bytes_le=b'\0'*15))
badvalue(lambda: self.uuid.UUID(bytes_le=b'\0'*17))
# Badly formed fields.
badvalue(lambda: self.uuid.UUID(fields=(1,)))
@@ -365,6 +408,70 @@ class BaseTestUUID:
badtype(lambda: u < object())
badtype(lambda: u > object())
def test_constructor_none(self):
# Explicit None has the same meaning as omitting an input argument.
expected = self.uuid.UUID('12345678-1234-5678-1234-567812345678')
inputs = {
'hex': expected.hex,
'bytes': expected.bytes,
'bytes_le': expected.bytes_le,
'fields': expected.fields,
'int': expected.int,
}
for name, value in inputs.items():
with self.subTest(input=name):
kwargs = dict.fromkeys(inputs)
kwargs[name] = value
self.assertEqual(self.uuid.UUID(**kwargs, version=None), expected)
self.assertEqual(
self.uuid.UUID(*kwargs.values(), version=None), expected)
with self.assertRaisesRegex(TypeError, 'one of the hex, bytes, bytes_le'):
self.uuid.UUID(**dict.fromkeys(inputs))
def test_constructor_bytes_type(self):
# A buffer interface alone does not satisfy the bytes arguments.
for name in ('bytes', 'bytes_le'):
for value in (bytearray(16), memoryview(bytes(16)), 'x' * 16,
[0] * 16, 16, object()):
with self.subTest(argument=name, type=type(value)):
with self.assertRaisesRegex(
TypeError, f'^{name} must be a bytes object$'
):
self.uuid.UUID(**{name: value})
def test_constructor_bytes_subclass(self):
class BytesSubclass(bytes):
pass
expected = self.uuid.UUID('12345678-1234-5678-1234-567812345678')
for name in ('bytes', 'bytes_le'):
with self.subTest(argument=name):
value = BytesSubclass(getattr(expected, name))
self.assertEqual(self.uuid.UUID(**{name: value}), expected)
def test_constructor_is_safe_none(self):
# is_safe is stored as supplied; None is distinct from SafeUUID.unknown.
self.assertIsNone(self.uuid.UUID(int=1, is_safe=None).is_safe)
def test_constructor_fields_length_error(self):
error = RuntimeError('fields length failed')
class Fields(tuple):
def __len__(self):
raise error
class NonSequence:
def __len__(self):
raise error
# Preserve __len__ errors, including for objects without __getitem__.
for fields in (Fields((0,) * 6), NonSequence()):
with self.subTest(type=type(fields)):
with self.assertRaises(RuntimeError) as cm:
self.uuid.UUID(fields=fields)
self.assertIs(cm.exception, error)
def test_getnode(self):
node1 = self.uuid.getnode()
self.assertTrue(0 < node1 < (1 << 48), '%012x' % node1)
@@ -388,6 +495,65 @@ class BaseTestUUID:
with self.subTest(protocol=proto):
check(pickle.loads(pickle.dumps(u, proto)), u)
def test_setstate_mapping(self):
# State restoration accepts mappings and decodes serialized safety values.
safety_states = [({}, self.uuid.SafeUUID.unknown)]
safety_states.extend(
({'is_safe': safety.value}, safety)
for safety in self.uuid.SafeUUID
)
with support.swap_item(sys.modules, 'uuid', self.uuid):
for mapping in (dict, frozendict, UserDict, MappingProxyType):
for safety_state, expected_safety in safety_states:
with self.subTest(mapping=mapping, state=safety_state):
u = self.uuid.UUID(int=0, is_safe=self.uuid.SafeUUID.safe)
state = mapping({'int': 1, **safety_state})
self.assertIsNone(u.__setstate__(state))
self.assertEqual(u.int, 1)
self.assertIs(u.is_safe, expected_safety)
def test_setstate_missing_int(self):
# A missing required key must retain the mapping's KeyError and key.
for mapping in (dict, frozendict, UserDict, MappingProxyType):
with self.subTest(mapping=mapping):
u = self.uuid.UUID(int=1)
with self.assertRaises(KeyError) as cm:
u.__setstate__(mapping({}))
self.assertEqual(cm.exception.args, ('int',))
self.assertEqual(u.int, 1)
def test_setstate_lookup_error(self):
class State(dict):
def __getitem__(self, key):
if key == bad_key:
raise RuntimeError('lookup failed')
return super().__getitem__(key)
# Honor overridden lookups, including on dict subclasses.
for bad_key in ('int', 'is_safe'):
with self.subTest(key=bad_key):
u = self.uuid.UUID(int=1)
state = State(int=2, is_safe=0)
with self.assertRaisesRegex(RuntimeError, 'lookup failed'):
u.__setstate__(state)
def test_setstate_contains_error(self):
class State(dict):
def __contains__(self, key):
raise RuntimeError('membership test failed')
# Errors checking for the optional key must propagate as well.
u = self.uuid.UUID(int=1)
state = State(int=2)
with self.assertRaisesRegex(RuntimeError, 'membership test failed'):
u.__setstate__(state)
def test_setstate_invalid_is_safe(self):
u = self.uuid.UUID(int=0)
with support.swap_item(sys.modules, 'uuid', self.uuid):
with self.assertRaisesRegex(ValueError, '42 is not a valid SafeUUID'):
u.__setstate__({'int': 1, 'is_safe': 42})
def test_unpickle_previous_python_versions(self):
def check(actual, expected):
self.assertEqual(actual, expected)
@@ -877,12 +1043,18 @@ class BaseTestUUID:
equal((u.int >> 80) & 0xffff, 0x232a)
equal((u.int >> 96) & 0xffff_ffff, 0x1ec9_414c)
def test_uuid7(self):
def test_uuid7_functional(self):
equal = self.assertEqual
u = self.uuid.uuid7()
equal(u.variant, self.uuid.RFC_4122)
equal(u.version, 7)
def test_uuid7_mock(self):
if self.is_c_uuid:
self.skipTest("C implementation of uuid7 cannot be tested with mocks")
equal = self.assertEqual
# 1 Jan 2023 12:34:56.123_456_789
timestamp_ns = 1672533296_123_456_789 # ns precision
timestamp_ms, _ = divmod(timestamp_ns, 1_000_000)
@@ -940,7 +1112,15 @@ class BaseTestUUID:
versions = {u.version for u in uuids}
self.assertSetEqual(versions, {7})
def test_uuid7_monotonicity(self):
def test_uuid7_monotonicity_functional(self):
equal = self.assertEqual
us = [self.uuid.uuid7() for _ in range(10_000)]
equal(us, sorted(us))
def test_uuid7_monotonicity_mock(self):
if self.is_c_uuid:
self.skipTest("C implementation of uuid7 cannot be tested with mocks")
equal = self.assertEqual
us = [self.uuid.uuid7() for _ in range(10_000)]
@@ -1003,7 +1183,10 @@ class BaseTestUUID:
self.assertLess(u1, u2)
def test_uuid7_timestamp_backwards(self):
def test_uuid7_timestamp_backwards_mock(self):
if self.is_c_uuid:
self.skipTest("C implementation of uuid7 cannot be tested with mocks")
equal = self.assertEqual
# 1 Jan 2023 12:34:56.123_456_789
timestamp_ns = 1672533296_123_456_789 # ns precision
@@ -1043,7 +1226,10 @@ class BaseTestUUID:
equal((u.int >> 32) & 0x3fff_ffff, counter_lo + 1)
equal(u.int & 0xffff_ffff, tail)
def test_uuid7_overflow_counter(self):
def test_uuid7_overflow_counter_mock(self):
if self.is_c_uuid:
self.skipTest("C implementation of uuid7 cannot be tested with mocks")
equal = self.assertEqual
# 1 Jan 2023 12:34:56.123_456_789
timestamp_ns = 1672533296_123_456_789 # ns precision
@@ -1149,6 +1335,7 @@ class BaseTestUUID:
class CommandLineTestCases:
uuid = None # to be defined in subclasses
is_c_uuid = False
def do_test_standalone_uuid(self, version):
stdout = io.StringIO()
@@ -1313,6 +1500,7 @@ class TestUUIDWithoutExtModule(CommandLineTestCases, BaseTestUUID, unittest.Test
@unittest.skipUnless(c_uuid, 'requires the C _uuid module')
class TestUUIDWithExtModule(CommandLineTestCases, BaseTestUUID, unittest.TestCase):
uuid = c_uuid
is_c_uuid = True
def check_has_stable_libuuid_extractable_node(self):
if not self.uuid._has_stable_extractable_node:
@@ -1343,6 +1531,7 @@ class TestUUIDWithExtModule(CommandLineTestCases, BaseTestUUID, unittest.TestCas
class BaseTestInternals:
_uuid = py_uuid
is_c_uuid = False
def check_parse_mac(self, aix):
if not aix:
@@ -1536,6 +1725,7 @@ class TestInternalsWithoutExtModule(BaseTestInternals, unittest.TestCase):
@unittest.skipUnless(c_uuid, 'requires the C _uuid module')
class TestInternalsWithExtModule(BaseTestInternals, unittest.TestCase):
uuid = c_uuid
is_c_uuid = True
@unittest.skipUnless(os.name == 'posix', 'requires Posix')
def test_unix_getnode(self):
@@ -1553,5 +1743,425 @@ class TestInternalsWithExtModule(BaseTestInternals, unittest.TestCase):
self.check_node(node)
class UuidHooks:
def __init__(self, *, start_at=0, inc_by=0, seed=0):
self._time = start_at
self._inc_by = inc_by
self._rnd = random.Random(seed)
def random_func(self, size):
ret = b''
for _ in range(size):
ret += self._rnd.getrandbits(8).to_bytes(1, 'big')
return ret
def time_func(self):
self._time += self._inc_by
return self._time
@unittest.skipUnless(c_uuid, "requires the C _uuid module")
class TestCImplementationCompat(unittest.TestCase):
def test_is_safe_lookup_error(self):
# Force the lazy SafeUUID lookup to fail before it is cached. The getter
# must propagate AttributeError and recover when the attribute returns.
# Run in a subprocess because using the failed lookup can crash Python.
assert_python_ok('-c', textwrap.dedent('''
import uuid
obj = uuid.UUID(int=1)
SafeUUID = uuid.SafeUUID
del uuid.SafeUUID
try:
try:
obj.is_safe
except AttributeError as exc:
assert 'SafeUUID' in str(exc)
else:
raise AssertionError('SafeUUID lookup unexpectedly succeeded')
finally:
uuid.SafeUUID = SafeUUID
assert obj.is_safe is SafeUUID.unknown
'''))
def test_compatibility(self):
import uuid
PU = uuid._py_UUID
CU = uuid._c_UUID
N = 1000
uuids = [
"00000000-0000-0000-0000-000000000000",
"ffffffff-ffff-ffff-ffff-ffffffffffff",
"c0bec4fd-e4e3-050c-a362-da3f734ffd56", # regression
*(str(uuid.uuid4()) for _ in range(N)),
*(str(uuid.uuid7()) for _ in range(N)),
*(str(uuid.uuid1()) for _ in range(N)),
*(str(uuid.UUID(bytes=os.urandom(16))) for _ in range(N)),
]
def full_test(p, u):
self.assertEqual(p, u)
self.assertEqual(p.hex, u.hex)
self.assertEqual(p.int, u.int)
self.assertEqual(p.variant, u.variant)
self.assertEqual(p.version, u.version)
self.assertEqual(p.is_safe, u.is_safe)
self.assertEqual(p.bytes, u.bytes)
self.assertEqual(p.bytes_le, u.bytes_le)
self.assertEqual(p.fields, u.fields)
self.assertEqual(p.time_low, u.time_low)
self.assertEqual(p.time_mid, u.time_mid)
self.assertEqual(p.time_hi_version, u.time_hi_version)
self.assertEqual(p.clock_seq_hi_variant, u.clock_seq_hi_variant)
self.assertEqual(p.clock_seq_low, u.clock_seq_low)
self.assertEqual(p.node, u.node)
all_ps = set()
all_us = set()
for uuid_str in uuids:
with self.subTest(uuid=uuid_str):
p = PU(uuid_str)
u = CU(uuid_str)
full_test(p, u)
u2 = CU(bytes_le=p.bytes_le)
full_test(p, u2)
u3 = CU(fields=p.fields)
full_test(p, u3)
u4 = CU(int=p.int)
full_test(p, u4)
u5 = CU(
hex=p.hex,
is_safe=uuid.SafeUUID.safe,
)
full_test(
PU(
uuid_str,
is_safe=uuid.SafeUUID.safe,
),
u5,
)
all_ps.add(p)
all_us.add(u)
self.assertEqual(len(all_ps), len(all_us))
self.assertEqual(len(all_ps), len(uuids))
def _install_hooks(self, uuid_mod, *, start_at=0, inc_by=0, seed=0):
py_hooks = UuidHooks(start_at=start_at, inc_by=inc_by, seed=seed)
uuid_mod._install_py_hooks(
random_func=py_hooks.random_func,
time_func=py_hooks.time_func
)
c_hooks = UuidHooks(start_at=start_at, inc_by=inc_by, seed=seed)
uuid_mod._install_c_hooks(
random_func=c_hooks.random_func,
time_func=c_hooks.time_func
)
def _reset_hooks(self, uuid_mod):
uuid_mod._install_c_hooks(random_func=None, time_func=None)
uuid_mod._install_py_hooks(random_func=None, time_func=None)
def test_invalid_time_hook_result(self):
import uuid
class IntSubclass(int):
pass
class Indexable:
def __index__(self):
raise AssertionError('invalid results must not be converted')
# Invalid results must be rejected before integer conversion.
for value in (None, False, 1.0, '1', object(),
IntSubclass(1), Indexable()):
with self.subTest(value=value):
uuid._install_c_hooks(random_func=None,
time_func=lambda: value)
try:
with self.assertRaisesRegex(ValueError,
'^time_func must return int$'):
uuid.uuid7()
finally:
uuid._install_c_hooks(random_func=None, time_func=None)
def test_time_hook_exception(self):
import uuid
def time_func():
raise RuntimeError('time hook failed')
uuid._install_c_hooks(random_func=None, time_func=time_func)
try:
with self.assertRaisesRegex(RuntimeError, 'time hook failed'):
uuid.uuid7()
finally:
uuid._install_c_hooks(random_func=None, time_func=None)
def test_exact_same_algo_uuid4(self):
import uuid
self._install_hooks(uuid)
try:
for seq_number in range(100):
with self.subTest(seq_number=seq_number):
self.assertEqual(
uuid._py_uuid4().hex,
uuid._c_uuid4().hex,
)
finally:
self._reset_hooks(uuid)
@threading_helper.requires_working_threading()
@threading_helper.reap_threads
def test_concurrent_entropy_refill(self):
# Concurrent calls must not consume the same cached entropy. Pause one
# UUIDv4 call during a refill after it consumes a partial cached block,
# then run another call. Each thread supplies its ID as entropy, making
# it possible to detect reuse of the first thread's cached prefix.
from concurrent.futures import ThreadPoolExecutor
import threading
import uuid
entered = threading.Event()
resume = threading.Event()
local = threading.local()
local.id = 1
local.pause_refill = False
def random_func(size):
local.refill_size = size
if local.pause_refill:
entered.set()
if not resume.wait(support.SHORT_TIMEOUT):
raise TimeoutError('entropy refill was not resumed')
return bytes([local.id]) * size
def generate(thread_id):
local.id = thread_id
local.pause_refill = thread_id == 1
return uuid.uuid4()
uuid._install_c_hooks(random_func=random_func, time_func=lambda: 0)
self.addCleanup(uuid._install_c_hooks, random_func=None, time_func=None)
# UUIDv7 consumes 10 bytes initially. Leave less than one UUIDv4's
# entropy cached, so the next call must combine cached and fresh bytes.
uuid.uuid7()
size = len(uuid.UUID(int=0).bytes)
cached = local.refill_size - 10
for _ in range(cached // size):
uuid.uuid4()
remaining = cached % size
self.assertGreater(remaining, 0)
with ThreadPoolExecutor(2) as pool:
pending = pool.submit(generate, 1)
try:
self.assertTrue(entered.wait(support.SHORT_TIMEOUT))
other = pool.submit(generate, 2).result(
timeout=support.SHORT_TIMEOUT)
finally:
resume.set()
first = pending.result(timeout=support.SHORT_TIMEOUT)
# The waiting call reserved the cached prefix; the other call must
# obtain its prefix from a new entropy block instead of reusing it.
self.assertNotEqual(first.bytes[:remaining], other.bytes[:remaining])
@threading_helper.requires_working_threading()
@threading_helper.reap_threads
def test_concurrent_uuid7_state(self):
# UUIDv7 must recheck its counter if a refill lets another call advance
# it. Pause one call in the entropy hook and complete another first;
# with fixed time and entropy, the resumed call needs a higher counter.
# Exercise both the first UUIDv7 call and an already initialized counter.
from concurrent.futures import ThreadPoolExecutor
import threading
import uuid
for initialized in (False, True):
with self.subTest(initialized=initialized):
entered = threading.Event()
resume = threading.Event()
calls = []
lock = threading.Lock()
blocked_call = 2 if initialized else 1
def random_func(size):
with lock:
calls.append(size)
call = len(calls)
if call == blocked_call:
entered.set()
if not resume.wait(support.SHORT_TIMEOUT):
raise TimeoutError('entropy refill was not resumed')
# Keep time and entropy identical, so only the shared
# counter can distinguish the concurrent UUIDs.
return bytes(size)
uuid._install_c_hooks(random_func=random_func,
time_func=lambda: 0)
try:
if initialized:
uuid.uuid7()
# Consume the remaining cache in four-byte tails,
# leaving the next UUIDv7 call to trigger a refill.
for _ in range((calls[0] - 10) // 4):
uuid.uuid7()
with ThreadPoolExecutor(1) as pool:
pending = pool.submit(uuid.uuid7)
try:
self.assertTrue(entered.wait(support.SHORT_TIMEOUT))
other = uuid.uuid7()
finally:
resume.set()
first = pending.result(timeout=support.SHORT_TIMEOUT)
# A call resumed after another UUID was generated must
# advance from that UUID's state, not its stale snapshot.
self.assertGreater(first, other)
finally:
uuid._install_c_hooks(random_func=None, time_func=None)
def test_exact_same_algo_uuid7(self):
import uuid
try:
for start_at, inc_by in [
(0, 0), (0, 10_000_000), (10_000_000 + 142, 1131827398127397)
]:
self._install_hooks(uuid, start_at=start_at, inc_by=inc_by)
for seq_number in range(100):
with self.subTest(
seq_number=seq_number, start_at=start_at, inc_by=inc_by,
):
self.assertEqual(
uuid._py_uuid7().hex,
uuid._c_uuid7().hex,
)
finally:
self._reset_hooks(uuid)
def test_subclassing(self):
import uuid
class U(uuid._c_UUID):
pass
u = U(int=1)
u_str = str(u)
del u
self.assertEqual(u_str, '00000000-0000-0000-0000-000000000001')
def test_subclass_allocation(self):
# Subclasses must bypass the base UUID freelist and get enough storage
# for their slots and dictionary, with GC tracking for reference cycles.
# Run in a subprocess because an incorrectly sized or reused object
# can corrupt memory before the assertions detect it.
assert_python_ok('-c', textwrap.dedent('''
import gc
import uuid
import weakref
class U(uuid.UUID):
__slots__ = ('extra', '__dict__')
# Exercise both the constructor and the internal factory.
for factory in (lambda cls: cls(int=1), lambda cls: cls._from_int(1)):
# Leave a base UUID available in the freelist.
base = uuid.UUID(int=0)
del base
u = factory(U)
# A subclass must not receive a cached base UUID.
assert type(u) is U
assert u.int == 1
# Allocation must include storage for subclass slots and a dict.
object.__setattr__(u, 'extra', u)
object.__setattr__(u, 'other', 42)
assert u.extra is u
assert u.other == 42
# The self-reference requires GC tracking to collect the object.
ref = weakref.ref(u)
del u
gc.collect()
assert ref() is None
# Subclass allocation must leave the base freelist usable.
base = uuid.UUID(int=2)
assert type(base) is uuid.UUID
assert base.int == 2
'''))
def test_subclass_module_finalization(self):
# GC may clear the base type's module reference before a subclass-owned
# instance is destroyed. Drop both modules and collect the subclass's
# class/instance cycle; deallocation must work without module state.
# Isolate the module removals and a possible crash in a subprocess.
assert_python_ok('-c', textwrap.dedent('''
import gc
import sys
import weakref
for from_int in (False, True):
import uuid
module = sys.modules['_uuid']
UUID = module.UUID
assert uuid.UUID is UUID
# Remove the module's UUID constants before creating the cycle.
del sys.modules['uuid']
del uuid
gc.collect()
class Sub(UUID):
pass
if from_int:
Sub.keep = Sub._from_int(1)
else:
Sub.keep = Sub(int=1)
refs = [weakref.ref(obj) for obj in
(module, UUID, Sub, Sub.keep)]
del sys.modules['_uuid']
del module, UUID, Sub
gc.collect()
assert all(ref() is None for ref in refs)
'''))
def test_freelist_type_lifetime(self):
# Reused live instances must keep their type and module alive, while
# cached instances must allow both to be collected.
module = import_helper.import_fresh_module('_uuid')
module_ref = weakref.ref(module)
type_ref = weakref.ref(module.UUID)
# Leave a base UUID available in the freelist.
obj = module.UUID(int=0)
del obj
# A reused live instance must still keep its type and module alive.
obj = module.UUID(int=1)
del module
support.gc_collect()
self.assertIsNotNone(type_ref())
self.assertIsNotNone(module_ref())
self.assertEqual(obj.int, 1)
del obj
support.gc_collect()
self.assertIsNone(type_ref())
self.assertIsNone(module_ref())
if __name__ == '__main__':
unittest.main()
Lib/uuid.py
+67 -21
View File
@@ -99,6 +99,7 @@ class SafeUUID:
_UINT_128_MAX = (1 << 128) - 1
# 128-bit mask to clear the variant and version bits of a UUID integral value
_RFC_4122_CLEARFLAGS_MASK = ~((0xf000 << 64) | (0xc000 << 48))
# RFC 4122 variant bits and version bits to activate on a UUID integral value.
@@ -110,6 +111,46 @@ _RFC_4122_VERSION_6_FLAGS = ((6 << 76) | (0x8000 << 48))
_RFC_4122_VERSION_7_FLAGS = ((7 << 76) | (0x8000 << 48))
_RFC_4122_VERSION_8_FLAGS = ((8 << 76) | (0x8000 << 48))
_random_hook = None
_time_hook = None
def _gen_random(size):
if _random_hook is None:
return os.urandom(size)
else:
return _random_hook(size)
def _gen_time():
if _time_hook is None:
return time.time_ns()
else:
return _time_hook()
def _install_py_hooks(*, random_func, time_func):
global _random_hook
global _time_hook
global _last_timestamp_v7
_random_hook = random_func
_time_hook = time_func
# Reset _last_timestamp_v7 for repeatability of tests
_last_timestamp_v7 = None
# Import optional C extension at toplevel, to help disabling it when testing
try:
import _uuid
_generate_time_safe = getattr(_uuid, "generate_time_safe", None)
_has_stable_extractable_node = _uuid.has_stable_extractable_node
_UuidCreate = getattr(_uuid, "UuidCreate", None)
except ImportError:
_uuid = None
_generate_time_safe = None
_has_stable_extractable_node = False
_UuidCreate = None
class UUID:
"""Instances of the UUID class represent UUIDs as specified in RFC 4122.
@@ -219,22 +260,24 @@ class UUID:
raise ValueError('badly formed hexadecimal UUID string')
int = int_(hex, 16)
elif bytes_le is not None:
if not isinstance(bytes_le, bytes_):
raise TypeError('bytes_le must be a bytes object')
if len(bytes_le) != 16:
raise ValueError('bytes_le is not a 16-char string')
assert isinstance(bytes_le, bytes_), repr(bytes_le)
bytes = (bytes_le[4-1::-1] + bytes_le[6-1:4-1:-1] +
bytes_le[8-1:6-1:-1] + bytes_le[8:])
int = int_.from_bytes(bytes) # big endian
elif bytes is not None:
if not isinstance(bytes, bytes_):
raise TypeError('bytes must be a bytes object')
if len(bytes) != 16:
raise ValueError('bytes is not a 16-char string')
assert isinstance(bytes, bytes_), repr(bytes)
int = int_.from_bytes(bytes) # big endian
elif fields is not None:
if len(fields) != 6:
raise ValueError('fields is not a 6-tuple')
(time_low, time_mid, time_hi_version,
clock_seq_hi_variant, clock_seq_low, node) = fields
clock_seq_hi_variant, clock_seq_low, node) = fields
if not 0 <= time_low < (1 << 32):
raise ValueError('field 1 out of range (need a 32-bit value)')
if not 0 <= time_mid < (1 << 16):
@@ -249,7 +292,7 @@ class UUID:
raise ValueError('field 6 out of range (need a 48-bit value)')
clock_seq = (clock_seq_hi_variant << 8) | clock_seq_low
int = ((time_low << 96) | (time_mid << 80) |
(time_hi_version << 64) | (clock_seq << 48) | node)
(time_hi_version << 64) | (clock_seq << 48) | node)
if not 0 <= int <= _UINT_128_MAX:
raise ValueError('int is out of range (need a 128-bit value)')
if version is not None:
@@ -629,19 +672,6 @@ def _netstat_getnode():
return _find_mac_under_heading('netstat', '-ian', b'Address')
# Import optional C extension at toplevel, to help disabling it when testing
try:
import _uuid
_generate_time_safe = getattr(_uuid, "generate_time_safe", None)
_has_stable_extractable_node = _uuid.has_stable_extractable_node
_UuidCreate = getattr(_uuid, "UuidCreate", None)
except ImportError:
_uuid = None
_generate_time_safe = None
_has_stable_extractable_node = False
_UuidCreate = None
def _unix_getnode():
"""Get the hardware address on Unix using the _uuid extension module."""
if _generate_time_safe and _has_stable_extractable_node:
@@ -774,7 +804,7 @@ def uuid3(namespace, name):
def uuid4():
"""Generate a random UUID."""
int_uuid_4 = int.from_bytes(os.urandom(16))
int_uuid_4 = int.from_bytes(_gen_random(16))
int_uuid_4 &= _RFC_4122_CLEARFLAGS_MASK
int_uuid_4 |= _RFC_4122_VERSION_4_FLAGS
return UUID._from_int(int_uuid_4)
@@ -834,7 +864,8 @@ _last_timestamp_v7 = None
_last_counter_v7 = 0 # 42-bit counter
def _uuid7_get_counter_and_tail():
rand = int.from_bytes(os.urandom(10))
rand = int.from_bytes(_gen_random(10))
# 42-bit counter with MSB set to 0
counter = (rand >> 32) & 0x1ff_ffff_ffff
# 32-bit random data
@@ -863,7 +894,7 @@ def uuid7():
global _last_timestamp_v7
global _last_counter_v7
nanoseconds = time.time_ns()
nanoseconds = _gen_time()
timestamp_ms = nanoseconds // 1_000_000
if _last_timestamp_v7 is None or timestamp_ms > _last_timestamp_v7:
@@ -879,7 +910,7 @@ def uuid7():
counter, tail = _uuid7_get_counter_and_tail()
else:
# 32-bit random data
tail = int.from_bytes(os.urandom(4))
tail = int.from_bytes(_gen_random(4))
unix_ts_ms = timestamp_ms & 0xffff_ffff_ffff
counter_msbs = counter >> 30
@@ -932,6 +963,21 @@ def uuid8(a=None, b=None, c=None):
return UUID._from_int(int_uuid_8)
_py_uuid4 = uuid4
_py_uuid7 = uuid7
_py_UUID = UUID
try:
from _uuid import UUID, uuid4, uuid7
from _uuid import _install_c_hooks as _install_c_hooks
except ImportError:
_c_UUID = None
_c_uuid4 = None
_c_uuid7 = None
else:
_c_UUID = UUID
_c_uuid4 = uuid4
_c_uuid7 = uuid7
def main():
"""Run the uuid command line interface."""
uuid_funcs = {
Misc/NEWS.d/next/Library/2025-09-18-14-13-00.gh-issue-139122.m3lp66.rst
+2
View File
@@ -0,0 +1,2 @@
:mod:`uuid`: Reimplement :class:`~uuid.UUID` objects, :func:`~uuid.uuid4`,
and :func:`~uuid.uuid7` in C.
Misc/NEWS.d/next/Library/2026-10-19-16-00-00.gh-issue-139122.bYtEs1.rst
+3
View File
@@ -0,0 +1,3 @@
:class:`uuid.UUID` now consistently raises :exc:`TypeError` when the *bytes*
or *bytes_le* argument is not a :class:`bytes` object. Previously, some invalid
inputs raised :exc:`AssertionError` or :exc:`ValueError`.
Modules/_uuidmodule.c
+1759 -9
View File
File diff suppressed because it is too large. Load diff
Modules/clinic/_uuidmodule.c.h
Generated
+299
View File
@@ -0,0 +1,299 @@
/*[clinic input]
preserve
[clinic start generated code]*/
#if defined(Py_BUILD_CORE) && !defined(Py_BUILD_CORE_MODULE)
# include "pycore_gc.h" // PyGC_Head
# include "pycore_runtime.h" // _Py_ID()
#endif
#include "pycore_critical_section.h"// Py_BEGIN_CRITICAL_SECTION()
#include "pycore_modsupport.h" // _PyArg_UnpackKeywords()
PyDoc_STRVAR(_uuid_uuid4__doc__,
"uuid4($module, /)\n"
"--\n"
"\n"
"Generate a random UUID (version 4).");
#define _UUID_UUID4_METHODDEF \
{"uuid4", (PyCFunction)_uuid_uuid4, METH_NOARGS, _uuid_uuid4__doc__},
static PyObject *
_uuid_uuid4_impl(PyObject *module);
static PyObject *
_uuid_uuid4(PyObject *module, PyObject *Py_UNUSED(ignored))
{
PyObject *return_value = NULL;
Py_BEGIN_CRITICAL_SECTION(module);
return_value = _uuid_uuid4_impl(module);
Py_END_CRITICAL_SECTION();
return return_value;
}
PyDoc_STRVAR(_uuid_uuid7__doc__,
"uuid7($module, /)\n"
"--\n"
"\n"
"Generate a UUID from a Unix timestamp in milliseconds and random bits.\n"
"\n"
"UUIDv7 objects feature monotonicity within a millisecond.");
#define _UUID_UUID7_METHODDEF \
{"uuid7", (PyCFunction)_uuid_uuid7, METH_NOARGS, _uuid_uuid7__doc__},
static PyObject *
_uuid_uuid7_impl(PyObject *module);
static PyObject *
_uuid_uuid7(PyObject *module, PyObject *Py_UNUSED(ignored))
{
PyObject *return_value = NULL;
Py_BEGIN_CRITICAL_SECTION(module);
return_value = _uuid_uuid7_impl(module);
Py_END_CRITICAL_SECTION();
return return_value;
}
PyDoc_STRVAR(_uuid__install_c_hooks__doc__,
"_install_c_hooks($module, /, *, random_func, time_func)\n"
"--\n"
"\n");
#define _UUID__INSTALL_C_HOOKS_METHODDEF \
{"_install_c_hooks", _PyCFunction_CAST(_uuid__install_c_hooks), METH_FASTCALL|METH_KEYWORDS, _uuid__install_c_hooks__doc__},
static PyObject *
_uuid__install_c_hooks_impl(PyObject *module, PyObject *random_func,
PyObject *time_func);
static PyObject *
_uuid__install_c_hooks(PyObject *module, PyObject *const *args, Py_ssize_t nargs, PyObject *kwnames)
{
PyObject *return_value = NULL;
#if defined(Py_BUILD_CORE) && !defined(Py_BUILD_CORE_MODULE)
#define NUM_KEYWORDS 2
static struct {
PyGC_Head _this_is_not_used;
PyObject_VAR_HEAD
Py_hash_t ob_hash;
PyObject *ob_item[NUM_KEYWORDS];
} _kwtuple = {
.ob_base = PyVarObject_HEAD_INIT(&PyTuple_Type, NUM_KEYWORDS)
.ob_hash = -1,
.ob_item = { &_Py_ID(random_func), &_Py_ID(time_func), },
};
#undef NUM_KEYWORDS
#define KWTUPLE (&_kwtuple.ob_base.ob_base)
#else // !Py_BUILD_CORE
# define KWTUPLE NULL
#endif // !Py_BUILD_CORE
static const char * const _keywords[] = {"random_func", "time_func", NULL};
static _PyArg_Parser _parser = {
.keywords = _keywords,
.fname = "_install_c_hooks",
.kwtuple = KWTUPLE,
};
#undef KWTUPLE
PyObject *argsbuf[2];
PyObject *random_func;
PyObject *time_func;
args = _PyArg_UnpackKeywords(args, nargs, NULL, kwnames, &_parser,
/*minpos*/ 0, /*maxpos*/ 0, /*minkw*/ 2, /*varpos*/ 0, argsbuf);
if (!args) {
goto exit;
}
random_func = args[0];
time_func = args[1];
Py_BEGIN_CRITICAL_SECTION(module);
return_value = _uuid__install_c_hooks_impl(module, random_func, time_func);
Py_END_CRITICAL_SECTION();
exit:
return return_value;
}
PyDoc_STRVAR(_uuid_UUID___init____doc__,
"UUID(hex=None, bytes=None, bytes_le=None, fields=None, int=None,\n"
" version=None, *, is_safe=<unrepresentable>)\n"
"--\n"
"\n"
"UUID is a fast base implementation type for uuid.UUID.");
static int
_uuid_UUID___init___impl(uuidobject *self, PyObject *hex, PyObject *bytes,
PyObject *bytes_le, PyObject *fields,
PyObject *int_value, PyObject *version,
PyObject *is_safe);
static int
_uuid_UUID___init__(PyObject *self, PyObject *args, PyObject *kwargs)
{
int return_value = -1;
#if defined(Py_BUILD_CORE) && !defined(Py_BUILD_CORE_MODULE)
#define NUM_KEYWORDS 7
static struct {
PyGC_Head _this_is_not_used;
PyObject_VAR_HEAD
Py_hash_t ob_hash;
PyObject *ob_item[NUM_KEYWORDS];
} _kwtuple = {
.ob_base = PyVarObject_HEAD_INIT(&PyTuple_Type, NUM_KEYWORDS)
.ob_hash = -1,
.ob_item = { &_Py_ID(hex), &_Py_ID(bytes), &_Py_ID(bytes_le), &_Py_ID(fields), &_Py_ID(int), &_Py_ID(version), &_Py_ID(is_safe), },
};
#undef NUM_KEYWORDS
#define KWTUPLE (&_kwtuple.ob_base.ob_base)
#else // !Py_BUILD_CORE
# define KWTUPLE NULL
#endif // !Py_BUILD_CORE
static const char * const _keywords[] = {"hex", "bytes", "bytes_le", "fields", "int", "version", "is_safe", NULL};
static _PyArg_Parser _parser = {
.keywords = _keywords,
.fname = "UUID",
.kwtuple = KWTUPLE,
};
#undef KWTUPLE
PyObject *argsbuf[7];
PyObject * const *fastargs;
Py_ssize_t nargs = PyTuple_GET_SIZE(args);
Py_ssize_t noptargs = nargs + (kwargs ? PyDict_GET_SIZE(kwargs) : 0) - 0;
PyObject *hex = Py_None;
PyObject *bytes = Py_None;
PyObject *bytes_le = Py_None;
PyObject *fields = Py_None;
PyObject *int_value = Py_None;
PyObject *version = Py_None;
PyObject *is_safe = NULL;
fastargs = _PyArg_UnpackKeywords(_PyTuple_CAST(args)->ob_item, nargs, kwargs, NULL, &_parser,
/*minpos*/ 0, /*maxpos*/ 6, /*minkw*/ 0, /*varpos*/ 0, argsbuf);
if (!fastargs) {
goto exit;
}
if (!noptargs) {
goto skip_optional_pos;
}
if (fastargs[0]) {
hex = fastargs[0];
if (!--noptargs) {
goto skip_optional_pos;
}
}
if (fastargs[1]) {
bytes = fastargs[1];
if (!--noptargs) {
goto skip_optional_pos;
}
}
if (fastargs[2]) {
bytes_le = fastargs[2];
if (!--noptargs) {
goto skip_optional_pos;
}
}
if (fastargs[3]) {
fields = fastargs[3];
if (!--noptargs) {
goto skip_optional_pos;
}
}
if (fastargs[4]) {
int_value = fastargs[4];
if (!--noptargs) {
goto skip_optional_pos;
}
}
if (fastargs[5]) {
version = fastargs[5];
if (!--noptargs) {
goto skip_optional_pos;
}
}
skip_optional_pos:
if (!noptargs) {
goto skip_optional_kwonly;
}
is_safe = fastargs[6];
skip_optional_kwonly:
return_value = _uuid_UUID___init___impl((uuidobject *)self, hex, bytes, bytes_le, fields, int_value, version, is_safe);
exit:
return return_value;
}
PyDoc_STRVAR(_uuid_UUID__from_int__doc__,
"_from_int($type, value, /)\n"
"--\n"
"\n"
"Create a UUID from an integer value. Internal use only.");
#define _UUID_UUID__FROM_INT_METHODDEF \
{"_from_int", (PyCFunction)_uuid_UUID__from_int, METH_O|METH_CLASS, _uuid_UUID__from_int__doc__},
static PyObject *
_uuid_UUID__from_int_impl(PyTypeObject *type, PyObject *value);
static PyObject *
_uuid_UUID__from_int(PyObject *type, PyObject *value)
{
PyObject *return_value = NULL;
return_value = _uuid_UUID__from_int_impl((PyTypeObject *)type, value);
return return_value;
}
PyDoc_STRVAR(_uuid_UUID___getstate____doc__,
"__getstate__($self, /)\n"
"--\n"
"\n"
"Return the UUID\'s state for pickling.");
#define _UUID_UUID___GETSTATE___METHODDEF \
{"__getstate__", (PyCFunction)_uuid_UUID___getstate__, METH_NOARGS, _uuid_UUID___getstate____doc__},
static PyObject *
_uuid_UUID___getstate___impl(uuidobject *self);
static PyObject *
_uuid_UUID___getstate__(PyObject *self, PyObject *Py_UNUSED(ignored))
{
return _uuid_UUID___getstate___impl((uuidobject *)self);
}
PyDoc_STRVAR(_uuid_UUID___setstate____doc__,
"__setstate__($self, state, /)\n"
"--\n"
"\n"
"Restore the UUID\'s state from pickling.\n"
"\n"
"Expects a mapping with \'int\' and optionally \'is_safe\' keys.");
#define _UUID_UUID___SETSTATE___METHODDEF \
{"__setstate__", (PyCFunction)_uuid_UUID___setstate__, METH_O, _uuid_UUID___setstate____doc__},
static PyObject *
_uuid_UUID___setstate___impl(uuidobject *self, PyObject *state);
static PyObject *
_uuid_UUID___setstate__(PyObject *self, PyObject *state)
{
PyObject *return_value = NULL;
return_value = _uuid_UUID___setstate___impl((uuidobject *)self, state);
return return_value;
}
/*[clinic end generated code: output=929e54bf30e649aa input=a9049054013a1b77]*/