Coverage for src/puuid/base.py: 87%

1"""

2pUUID base implementation.

4Provides the abstract base class and version-specific implementations for prefixed UUIDs.

5"""

7import annotationlib

8from abc import ABC, abstractmethod

9from typing import (

10 TYPE_CHECKING,

11 Any,

12 ClassVar,

13 Final,

14 Literal,

15 Self,

16 TypeAliasType,

17 final,

18 get_args,

19 get_origin,

20 override,

21)

22from uuid import NAMESPACE_DNS, UUID, uuid1, uuid3, uuid4, uuid5, uuid6, uuid7, uuid8

24if TYPE_CHECKING:

25 from pydantic import GetCoreSchemaHandler, GetJsonSchemaHandler

26 from pydantic.json_schema import JsonSchemaValue

27 from pydantic_core import core_schema

29 _PYDANTIC_AVAILABLE = True

30else:

31 try:

32 from pydantic import GetCoreSchemaHandler, GetJsonSchemaHandler

33 from pydantic.json_schema import JsonSchemaValue

34 from pydantic_core import core_schema

36 _PYDANTIC_AVAILABLE = True

37 except ModuleNotFoundError:

38 _PYDANTIC_AVAILABLE = False

40 class GetCoreSchemaHandler: ...

42 class _CoreSchema: ...

44 @final

45 class core_schema:

46 CoreSchema = _CoreSchema

49@final

50class ERR_MSG:

51 UUID_VERSION_MISMATCH = "Expected 'UUID' with version '{expected}', got '{actual}'"

52 PREFIX_DESERIALIZATION_ERROR = "Unable to deserialize prefix '{prefix}', separator '_' or UUID for '{classname}' from '{serial_puuid}'!"

53 INVALID_TYPE_FOR_SERIAL_PUUID = "'{classname}' can not be created from invalid type '{type}' with value '{value}'!"

54 EMPTY_PREFIX_DISALLOWED = "Empty prefix is not allowed for '{classname}'!"

57class PUUIDError(Exception):

58 """Base exception for pUUID related errors."""

60 message: str

62 def __init__(self, message: str = "") -> None:

63 super().__init__(message)

64 self.message = message

67################################################################################

68#### utilities

69################################################################################

72def _evaluate_type_alias(value: object) -> object:

73 """

74 Resolve a PEP 695 `type Alias = ...` (TypeAliasType) to its underlying value.

75 """

76 if isinstance(value, TypeAliasType): 76 ↛ 77line 76 didn't jump to line 77 because the condition on line 76 was never true

77 return annotationlib.call_evaluate_function(

78 value.evaluate_value,

79 annotationlib.Format.VALUE,

80 owner=value,

81 )

82 return value

85def _try_extract_literal_string(item: object) -> str | None:

86 """

87 If `item` is `Literal["..."]` (possibly via a `type` alias), return its string.

88 Otherwise return None.

89 """

90 evaluated = _evaluate_type_alias(item)

91 if get_origin(evaluated) is Literal:

92 args = get_args(evaluated)

93 if len(args) == 1 and isinstance(args[0], str): 93 ↛ 95line 93 didn't jump to line 95 because the condition on line 93 was always true

94 return args[0]

95 return None

98################################################################################

99#### PUUIDBase

100################################################################################

101

102

103class PUUIDBase[TPrefix: str](ABC):

104 """Abstract Generic Base Class for Prefixed UUIDs."""

105

106 _prefix: ClassVar[str] = ""

107 _serial: str | None = None

108 _uuid: UUID

109

110 def __init_subclass__(cls, **kwargs: Any) -> None:

111 super().__init_subclass__(**kwargs)

112 # Automatic prefix assignment from the type parameter

113 for base in getattr(cls, "__orig_bases__", []): 113 ↛ 129line 113 didn't jump to line 129 because the loop on line 113 didn't complete

114 origin = get_origin(base)

115 if origin is not None and issubclass(origin, PUUIDBase): 115 ↛ 113line 115 didn't jump to line 113 because the condition on line 115 was always true

116 args = get_args(base)

117 if args: 117 ↛ 113line 117 didn't jump to line 113 because the condition on line 117 was always true

118 prefix = _try_extract_literal_string(args[0])

119 if prefix is None: # e.g. args[0] is TypeVar TPrefix

120 return

121 if prefix == "":

122 raise PUUIDError(

123 ERR_MSG.EMPTY_PREFIX_DISALLOWED.format(

124 classname=cls.__name__

125 )

126 )

127 cls._prefix = prefix

128 return

129 raise AssertionError(

130 "Something unexpected happened in the usage of the PUUID library"

131 )

132

133 @abstractmethod

134 def __init__(self, uuid: UUID, *, check_version: bool = True) -> None: ...

135

136 def __new__(cls, *args: Any, **kwargs: Any) -> Self:

137 instance = super().__new__(cls)

138 if not cls._prefix:

139 raise PUUIDError(

140 ERR_MSG.EMPTY_PREFIX_DISALLOWED.format(classname=cls.__name__)

141 )

142 return instance

143

144 @classmethod

145 def prefix(cls) -> str:

146 """

147 Return the defined prefix for the class.

148

149 Returns

150 -------

151 str

152 The prefix string.

153 """

154 return cls._prefix

155

156 @property

157 def uuid(self) -> UUID:

158 """

159 Return the underlying UUID object.

160

161 Returns

162 -------

163 UUID

164 The native UUID instance.

165 """

166 return self._uuid

167

168 def _format_serial(self) -> str:

169 return f"{type(self)._prefix}_{self._uuid}"

170

171 def to_string(self) -> str:

172 """

173 Return the string representation of the Prefixed UUID.

174

175 Returns

176 -------

177 str

178 The formatted string (e.g., `<prefix>_<uuid-hex-string>`).

179 """

180 if self._serial is not None:

181 return self._serial

182

183 serial = self._format_serial()

184 self._serial = serial

185 return serial

186

187 @classmethod

188 def from_string(cls, serial_puuid: str) -> Self:

189 """

190 Create a pUUID instance from its string representation.

191

192 Parameters

193 ----------

194 serial_puuid : str

195 The prefixed UUID string (e.g., `user_550e8400-e29b...`).

196

197 Returns

198 -------

199 Self

200 The deserialized pUUID instance.

201

202 Raises

203 ------

204 PUUIDError

205 If the string is malformed or the prefix does not match.

206 """

207 try:

208 if "_" not in serial_puuid:

209 raise ValueError("Missing separator")

210

211 prefix, serialized_uuid = serial_puuid.split("_", 1)

212

213 if prefix != cls._prefix:

214 raise ValueError("Prefix mismatch")

215

216 uuid = UUID(serialized_uuid)

217 return cls(uuid=uuid)

218

219 except ValueError as err:

220 raise PUUIDError(

221 ERR_MSG.PREFIX_DESERIALIZATION_ERROR.format(

222 prefix=cls._prefix,

223 classname=cls.__name__,

224 serial_puuid=serial_puuid,

225 )

226 ) from err

227

228 @override

229 def __str__(self) -> str:

230 return self.to_string()

231

232 @override

233 def __eq__(self, other: object) -> bool:

234 if not isinstance(other, PUUIDBase):

235 return False

236

237 return (self._prefix, self._uuid) == (other._prefix, other._uuid)

238

239 @override

240 def __hash__(self) -> int:

241 return hash((type(self)._prefix, self._uuid))

242

243 @classmethod

244 def __get_pydantic_core_schema__(

245 cls,

246 _source_type: object,

247 _handler: GetCoreSchemaHandler,

248 ) -> core_schema.CoreSchema:

249 if not _PYDANTIC_AVAILABLE:

250 raise ModuleNotFoundError(

251 "pydantic is an optional dependency. Install with: pip install 'pUUID[pydantic]'"

252 )

253

254 def validate(value: object) -> PUUIDBase[TPrefix]:

255 if isinstance(value, cls):

256 return value

257

258 if isinstance(value, str):

259 try:

260 return cls.from_string(value)

261 except PUUIDError as err:

262 raise ValueError(str(err)) from err

263

264 raise ValueError(

265 ERR_MSG.INVALID_TYPE_FOR_SERIAL_PUUID.format(

266 classname=cls.__name__, type=type(value), value=value

267 )

268 )

269

270 def serialize(value: PUUIDBase[TPrefix]) -> str:

271 return value.to_string()

272

273 def wrap_validate(value: object, handler: Any) -> PUUIDBase[TPrefix]:

274 return validate(value)

275

276 return core_schema.json_or_python_schema(

277 json_schema=core_schema.no_info_wrap_validator_function(

278 wrap_validate,

279 core_schema.str_schema(),

280 ),

281 python_schema=core_schema.no_info_plain_validator_function(validate),

282 serialization=core_schema.plain_serializer_function_ser_schema(

283 serialize,

284 return_schema=core_schema.str_schema(),

285 ),

286 )

287

288 @classmethod

289 def __get_pydantic_json_schema__(

290 cls, core_schema: core_schema.CoreSchema, handler: GetJsonSchemaHandler

291 ) -> JsonSchemaValue:

292 version = getattr(cls, "VERSION", 0)

293 match version:

294 case 1: 294 ↛ 295line 294 didn't jump to line 295 because the pattern on line 294 never matched

295 examples = [f"{cls._prefix}_{uuid1()}" for _ in range(3)]

296 case 3: 296 ↛ 297line 296 didn't jump to line 297 because the pattern on line 296 never matched

297 examples = [

298 f"{cls._prefix}_{uuid3(namespace=NAMESPACE_DNS, name="digon.io")}"

299 for _ in range(3)

300 ]

301 case 4:

302 examples = [f"{cls._prefix}_{uuid4()}" for _ in range(3)]

303 case 5: 303 ↛ 304line 303 didn't jump to line 304 because the pattern on line 303 never matched

304 examples = [

305 f"{cls._prefix}_{uuid5(namespace=NAMESPACE_DNS, name="digon.io")}"

306 for _ in range(3)

307 ]

308 case 6: 308 ↛ 309line 308 didn't jump to line 309 because the pattern on line 308 never matched

309 examples = [f"{cls._prefix}_{uuid6()}" for _ in range(3)]

310 case 7: 310 ↛ 312line 310 didn't jump to line 312 because the pattern on line 310 always matched

311 examples = [f"{cls._prefix}_{uuid7()}" for _ in range(3)]

312 case 8:

313 examples = [f"{cls._prefix}_{uuid8()}" for _ in range(3)]

314 case _:

315 raise PUUIDError()

316

317 return {

318 "type": "string",

319 "title": cls.__name__,

320 "description": f"Prefixed UUID with prefix '{cls._prefix}'",

321 "examples": examples,

322 "pattern": rf"^{cls._prefix}_[0-9a-fA-F-]{{36}}$",

323 }

324

325

326################################################################################

327#### PUUIDv1

328################################################################################

329

330

331class PUUIDv1[TPrefix: str](PUUIDBase[TPrefix]):

332 """Prefixed UUID Version 1 (MAC address and time)."""

333

334 VERSION: Final[int] = 1

335

336 def __init__(

337 self,

338 uuid: UUID,

339 *,

340 check_version: bool = True,

341 ) -> None:

342 """

343 Initialize a PUUIDv1.

344

345 Parameters

346 ----------

347 uuid : UUID

348 An UUIDv1 instance.

349

350 Raises

351 ------

352 PUUIDError

353 If the UUID version is incorrect.

354 """

355

356 if check_version and (uuid.version != 1):

357 raise PUUIDError(

358 ERR_MSG.UUID_VERSION_MISMATCH.format(expected=1, actual=uuid.version)

359 )

360 self._uuid = uuid

361

362 @classmethod

363 def factory(

364 cls,

365 *,

366 node: int | None = None,

367 clock_seq: int | None = None,

368 ) -> Self:

369 """

370 Create a new PUUIDv1 instance. using current time and MAC address.

371

372 Parameters

373 ----------

374 node : int | None, optional

375 MAC address.

376 clock_seq : int | None, optional

377 The current time.

378

379 Returns

380 -------

381 Self

382 A new PUUIDv1 instance.

383 """

384

385 return cls(uuid1(node, clock_seq), check_version=False)

386

387

388################################################################################

389#### PUUIDv3

390################################################################################

391

392

393class PUUIDv3[TPrefix: str](PUUIDBase[TPrefix]):

394 """Prefixed UUID Version 3 (MD5 hash of namespace and name)."""

395

396 VERSION: Final[int] = 3

397

398 def __init__(

399 self,

400 uuid: UUID,

401 *,

402 check_version: bool = True,

403 ) -> None:

404 """

405 Initialize a PUUIDv3.

406

407 Parameters

408 ----------

409 uuid : UUID

410 An UUIDv3 instance.

411

412 Raises

413 ------

414 PUUIDError

415 If the UUID version is incorrect.

416 """

417

418 if check_version and (uuid.version != 3):

419 raise PUUIDError(

420 ERR_MSG.UUID_VERSION_MISMATCH.format(expected=3, actual=uuid.version)

421 )

422 self._uuid = uuid

423

424 @classmethod

425 def factory(cls, *, namespace: UUID, name: str | bytes) -> Self:

426 """

427 Create a new PUUIDv3.

428

429 Parameters

430 ----------

431 namespace : UUID | None, optional

432 Namespace UUID.

433 name : str | bytes | None, optional

434 The name used for hashing.

435

436 Returns

437 -------

438 Self

439 A new PUUIDv3 instance.

440 """

441

442 return cls(uuid3(namespace, name), check_version=False)

443

444

445################################################################################

446#### PUUIDv4

447################################################################################

448

449

450class PUUIDv4[TPrefix: str](PUUIDBase[TPrefix]):

451 """Prefixed UUID Version 4 (randomly generated)."""

452

453 VERSION: Final[int] = 4

454

455 def __init__(

456 self,

457 uuid: UUID,

458 *,

459 check_version: bool = True,

460 ) -> None:

461 """

462 Initialize a PUUIDv4.

463

464 Parameters

465 ----------

466 uuid : UUID

467 An UUIDv4 instance.

468

469 Raises

470 ------

471 PUUIDError

472 If the UUID version is incorrect.

473 """

474

475 if check_version and (uuid.version != 4):

476 raise PUUIDError(

477 ERR_MSG.UUID_VERSION_MISMATCH.format(expected=4, actual=uuid.version)

478 )

479 self._uuid = uuid

480

481 @classmethod

482 def factory(cls) -> Self:

483 """

484 Create a new PUUIDv4 instance using random generation.

485

486 Returns

487 -------

488 Self

489 A new PUUIDv4 instance.

490 """

491

492 return cls(uuid4(), check_version=False)

493

494

495################################################################################

496#### PUUIDv5

497################################################################################

498

499

500class PUUIDv5[TPrefix: str](PUUIDBase[TPrefix]):

501 """Prefixed UUID Version 5 (SHA-1 hash of namespace and name)."""

502

503 VERSION: Final[int] = 5

504

505 def __init__(

506 self,

507 uuid: UUID,

508 *,

509 check_version: bool = True,

510 ) -> None:

511 """

512 Initialize a PUUIDv5.

513

514 Parameters

515 ----------

516 uuid : UUID

517 Existing UUIDv5 instance.

518

519 Raises

520 ------

521 PUUIDError

522 If the UUID version is incorrect.

523 """

524

525 if check_version and (uuid.version != 5):

526 raise PUUIDError(

527 ERR_MSG.UUID_VERSION_MISMATCH.format(expected=5, actual=uuid.version)

528 )

529 self._uuid = uuid

530

531 @classmethod

532 def factory(

533 cls,

534 *,

535 namespace: UUID,

536 name: str | bytes,

537 ) -> Self:

538 """

539 Create a new PUUIDv5 instance using random generation.

540

541 Parameters

542 ----------

543 namespace : UUID

544 Namespace UUID.

545 name : str | bytes

546 The name used for hashing.

547

548 Returns

549 -------

550 Self

551 A new PUUIDv5 instance.

552 """

553

554 return cls(uuid5(namespace=namespace, name=name), check_version=False)

555

556

557################################################################################

558#### PUUIDv6

559################################################################################

560

561

562class PUUIDv6[TPrefix: str](PUUIDBase[TPrefix]):

563 """Prefixed UUID Version 6 (reordered v1 for DB locality)."""

564

565 VERSION: Final[int] = 6

566

567 def __init__(

568 self,

569 uuid: UUID,

570 *,

571 check_version: bool = True,

572 ) -> None:

573 """

574 Initialize a PUUIDv6.

575

576 Parameters

577 ----------

578 uuid : UUID

579 An UUIDv6 instance.

580

581 Raises

582 ------

583 PUUIDError

584 If the UUID version is incorrect.

585 """

586

587 if check_version and (uuid.version != 6):

588 raise PUUIDError(

589 ERR_MSG.UUID_VERSION_MISMATCH.format(expected=6, actual=uuid.version)

590 )

591 self._uuid = uuid

592

593 @classmethod

594 def factory(

595 cls,

596 *,

597 node: int | None = None,

598 clock_seq: int | None = None,

599 ) -> Self:

600 """

601 Create a new PUUIDv6 instance. using current time and MAC address.

602

603 Parameters

604 ----------

605 node : int | None, optional

606 MAC address.

607 clock_seq : int | None, optional

608 The current time.

609

610 Returns

611 -------

612 Self

613 A new PUUIDv6 instance.

614 """

615

616 return cls(uuid6(node, clock_seq), check_version=False)

617

618

619################################################################################

620#### PUUIDv7

621################################################################################

622

623

624class PUUIDv7[TPrefix: str](PUUIDBase[TPrefix]):

625 """Prefixed UUID Version 7 (time-ordered)."""

626

627 VERSION: Final[int] = 7

628

629 def __init__(

630 self,

631 uuid: UUID,

632 *,

633 check_version: bool = True,

634 ) -> None:

635 """

636 Initialize a PUUIDv7.

637

638 Parameters

639 ----------

640 uuid : UUID

641 An UUIDv7 instance.

642

643 Raises

644 ------

645 PUUIDError

646 If the UUID version is incorrect.

647 """

648

649 if check_version and (uuid.version != 7):

650 raise PUUIDError(

651 ERR_MSG.UUID_VERSION_MISMATCH.format(expected=7, actual=uuid.version)

652 )

653 self._uuid = uuid

654

655 @classmethod

656 def factory(cls) -> Self:

657 """

658 Create a new PUUIDv7 instance using random generation.

659

660 Returns

661 -------

662 Self

663 A new PUUIDv7 instance.

664 """

665

666 return cls(uuid7(), check_version=False)

667

668

669################################################################################

670#### PUUIDv8

671################################################################################

672

673

674class PUUIDv8[TPrefix: str](PUUIDBase[TPrefix]):

675 """Prefixed UUID Version 8 (custom implementation)."""

676

677 VERSION: Final[int] = 8

678

679 def __init__(

680 self,

681 uuid: UUID,

682 *,

683 check_version: bool = True,

684 ) -> None:

685 """

686 Initialize a PUUIDv8.

687

688 Parameters

689 ----------

690 uuid : UUID

691 An UUIDv8 instance.

692

693 Raises

694 ------

695 PUUIDError

696 If the UUID version is incorrect.

697 """

698

699 if check_version and (uuid.version != 8):

700 raise PUUIDError(

701 ERR_MSG.UUID_VERSION_MISMATCH.format(expected=8, actual=uuid.version)

702 )

703 self._uuid = uuid

704

705 @classmethod

706 def factory(

707 cls,

708 *,

709 a: int | None = None,

710 b: int | None = None,

711 c: int | None = None,

712 ) -> Self:

713 """

714 Create a new PUUIDv8 instance using custom generation.

715

716 Parameters

717 ----------

718 a : int | None, optional

719 First custom 48-bit value.

720 b : int | None, optional

721 Second custom 12-bit value.

722 c : int | None, optional

723 Third custom 62-bit value.

724

725 Returns

726 -------

727 Self

728 A new PUUIDv8 instance.

729 """

730

731 return cls(uuid8(a, b, c), check_version=False)

Coverage for src / puuid / base.py: 87%

199 statements