uri-register 0.3.0

A high-performance PostgreSQL-backed URI dictionary service for assigning unique integer IDs to URIs
Documentation 
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

"""Type stubs for uri_register"""

__version__: str

class UriRegister:
    """
    A high-performance PostgreSQL-backed URI to ID mapping service.

    This class provides async methods to register URIs and retrieve their
    unique integer IDs. All operations are atomic and thread-safe.
    """

    @staticmethod
    async def new(
        database_url: str, max_connections: int, cache_size: int
    ) -> UriRegister:
        """
        Create a new URI register connected to PostgreSQL.

        Args:
            database_url: PostgreSQL connection string (e.g., "postgres://user:password@host:port/database")
            max_connections: Maximum number of connections in the pool (recommended: 10-50)
            cache_size: Number of URI-to-ID mappings to cache in memory (recommended: 1,000-100,000)

        Returns:
            A new UriRegister instance

        Raises:
            RuntimeError: If connection to database fails

        Example:
            >>> register = await UriRegister.new("postgres://localhost/mydb", 20, 10000)
        """
        ...
    async def register_uri(self, uri: str) -> int:
        """
        Register a single URI and return its unique ID.

        If the URI already exists, returns its existing ID.
        If the URI is new, assigns and returns a new ID.

        Args:
            uri: The URI string to register

        Returns:
            The unique integer ID for this URI

        Raises:
            RuntimeError: If registration fails

        Example:
            >>> id = await register.register_uri("http://example.org")
            >>> print(f"Registered with ID: {id}")
        """
        ...
    async def register_uri_batch(self, uris: list[str]) -> list[int]:
        """
        Register multiple URIs in batch and return their IDs.

        Order is preserved: ids[i] corresponds to uris[i].
        This is much faster than calling register_uri() in a loop.

        Args:
            uris: List of URI strings to register

        Returns:
            List of integer IDs in the same order as input URIs

        Raises:
            RuntimeError: If batch registration fails

        Example:
            >>> uris = ["http://example.org/1", "http://example.org/2"]
            >>> ids = await register.register_uri_batch(uris)
            >>> # ids[0] corresponds to uris[0], ids[1] to uris[1]
        """
        ...
    async def register_uri_batch_hashmap(self, uris: list[str]) -> dict[str, int]:
        """
        Register multiple URIs in batch and return a dict mapping URIs to IDs.

        Duplicate URIs in the input are automatically deduplicated.

        Args:
            uris: List of URI strings to register

        Returns:
            Dictionary mapping each unique URI to its integer ID

        Raises:
            RuntimeError: If batch registration fails

        Example:
            >>> uris = ["http://example.org/1", "http://example.org/2"]
            >>> mapping = await register.register_uri_batch_hashmap(uris)
            >>> id1 = mapping["http://example.org/1"]
        """
        ...
    async def stats(self) -> dict[str, int]:
        """
        Get statistics about the URI register.

        Returns:
            Dictionary with keys:
            - 'total_uris': Total number of unique URIs registered
            - 'size_bytes': Total storage size in bytes (includes indexes)

        Raises:
            RuntimeError: If stats query fails

        Example:
            >>> stats = await register.stats()
            >>> print(f"Total URIs: {stats['total_uris']}")
            >>> print(f"Size: {stats['size_bytes']} bytes")
        """
        ...

__all__ = ["UriRegister"]