Skip to content

domain.protocols.user_repository

src.domain.protocols.user_repository

UserRepository protocol for user persistence.

Port (interface) for hexagonal architecture. Infrastructure layer implements this protocol.

Classes

UserRepository

Bases: Protocol

User repository protocol (port).

Defines the interface for user persistence operations. Infrastructure layer provides concrete implementation.

This is a Protocol (not ABC) for structural typing. Implementations don't need to inherit from this.

Methods:

Name Description
find_by_id

Retrieve user by ID
find_by_email

Retrieve user by email
save

Create new user
update

Update existing user
delete

Delete user (soft delete)
Example Implementation

class UserRepository: ... async def find_by_email(self, email: str) -> User | None: ... # Database logic here ... pass

Source code in src/domain/protocols/user_repository.py 
 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
class UserRepository(Protocol):
    """User repository protocol (port).

    Defines the interface for user persistence operations.
    Infrastructure layer provides concrete implementation.

    This is a Protocol (not ABC) for structural typing.
    Implementations don't need to inherit from this.

    Methods:
        find_by_id: Retrieve user by ID
        find_by_email: Retrieve user by email
        save: Create new user
        update: Update existing user
        delete: Delete user (soft delete)

    Example Implementation:
        >>> class UserRepository:
        ...     async def find_by_email(self, email: str) -> User | None:
        ...         # Database logic here
        ...         pass
    """

    async def find_by_id(self, user_id: UUID) -> User | None:
        """Find user by ID.

        Args:
            user_id: User's unique identifier.

        Returns:
            User if found, None otherwise.

        Example:
            >>> user = await repo.find_by_id(uuid7())
            >>> if user:
            ...     print(user.email)
        """
        ...

    async def find_by_email(self, email: str) -> User | None:
        """Find user by email address.

        Email comparison should be case-insensitive.

        Args:
            email: User's email address (case-insensitive).

        Returns:
            User if found, None otherwise.

        Example:
            >>> user = await repo.find_by_email("user@example.com")
            >>> if user:
            ...     print(user.id)
        """
        ...

    async def save(self, user: User) -> None:
        """Create new user in database.

        Args:
            user: User entity to persist.

        Raises:
            ConflictError: If email already exists.
            DatabaseError: If database operation fails.

        Example:
            >>> user = User(id=uuid7(), email="new@example.com", ...)
            >>> await repo.save(user)
        """
        ...

    async def update(self, user: User) -> None:
        """Update existing user in database.

        Args:
            user: User entity with updated fields.

        Raises:
            NotFoundError: If user doesn't exist.
            DatabaseError: If database operation fails.

        Example:
            >>> user.failed_login_attempts = 0
            >>> await repo.update(user)
        """
        ...

    async def delete(self, user_id: UUID) -> None:
        """Delete user (soft delete - sets is_active=False).

        Args:
            user_id: User's unique identifier.

        Raises:
            NotFoundError: If user doesn't exist.
            DatabaseError: If database operation fails.

        Note:
            This is a soft delete. User remains in database but is_active=False.

        Example:
            >>> await repo.delete(user_id)
        """
        ...

    async def exists_by_email(self, email: str) -> bool:
        """Check if user with email exists.

        Convenience method to check for duplicates before creation.

        Args:
            email: Email address to check (case-insensitive).

        Returns:
            True if user exists, False otherwise.

        Example:
            >>> if await repo.exists_by_email("user@example.com"):
            ...     raise ConflictError("Email already exists")
        """
        ...

    async def update_password(self, user_id: UUID, password_hash: str) -> None:
        """Update user's password hash.

        Atomic operation that only updates the password field.
        Used for password reset and password change flows.

        Args:
            user_id: User's unique identifier.
            password_hash: New bcrypt password hash.

        Raises:
            NotFoundError: If user doesn't exist.
            DatabaseError: If database operation fails.

        Example:
            >>> await repo.update_password(user_id, new_hash)
        """
        ...

    async def verify_email(self, user_id: UUID) -> None:
        """Mark user's email as verified.

        Atomic operation that sets is_verified=True.
        Used for email verification flow.

        Args:
            user_id: User's unique identifier.

        Raises:
            NotFoundError: If user doesn't exist.
            DatabaseError: If database operation fails.

        Example:
            >>> await repo.verify_email(user_id)
        """
        ...
Functions
find_by_id async 
find_by_id(user_id: UUID) -> User | None

Find user by ID.

Parameters:

Name Type Description Default
user_id UUID

User's unique identifier.

 required

Returns:

Type Description
User | None

User if found, None otherwise.
Example

user = await repo.find_by_id(uuid7()) if user: ... print(user.email)

Source code in src/domain/protocols/user_repository.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
async def find_by_id(self, user_id: UUID) -> User | None:
    """Find user by ID.

    Args:
        user_id: User's unique identifier.

    Returns:
        User if found, None otherwise.

    Example:
        >>> user = await repo.find_by_id(uuid7())
        >>> if user:
        ...     print(user.email)
    """
    ...
find_by_email async 
find_by_email(email: str) -> User | None

Find user by email address.

Email comparison should be case-insensitive.

Parameters:

Name Type Description Default
email str

User's email address (case-insensitive).

 required

Returns:

Type Description
User | None

User if found, None otherwise.
Example

user = await repo.find_by_email("user@example.com") if user: ... print(user.id)

Source code in src/domain/protocols/user_repository.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
async def find_by_email(self, email: str) -> User | None:
    """Find user by email address.

    Email comparison should be case-insensitive.

    Args:
        email: User's email address (case-insensitive).

    Returns:
        User if found, None otherwise.

    Example:
        >>> user = await repo.find_by_email("user@example.com")
        >>> if user:
        ...     print(user.id)
    """
    ...
save async 
save(user: User) -> None

Create new user in database.

Parameters:

Name Type Description Default
user User

User entity to persist.

 required

Raises:

Type Description
ConflictError

If email already exists.
DatabaseError

If database operation fails.
Example

user = User(id=uuid7(), email="new@example.com", ...) await repo.save(user)

Source code in src/domain/protocols/user_repository.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
async def save(self, user: User) -> None:
    """Create new user in database.

    Args:
        user: User entity to persist.

    Raises:
        ConflictError: If email already exists.
        DatabaseError: If database operation fails.

    Example:
        >>> user = User(id=uuid7(), email="new@example.com", ...)
        >>> await repo.save(user)
    """
    ...
update async 
update(user: User) -> None

Update existing user in database.

Parameters:

Name Type Description Default
user User

User entity with updated fields.

 required

Raises:

Type Description
NotFoundError

If user doesn't exist.
DatabaseError

If database operation fails.
Example

user.failed_login_attempts = 0 await repo.update(user)

Source code in src/domain/protocols/user_repository.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def update(self, user: User) -> None:
    """Update existing user in database.

    Args:
        user: User entity with updated fields.

    Raises:
        NotFoundError: If user doesn't exist.
        DatabaseError: If database operation fails.

    Example:
        >>> user.failed_login_attempts = 0
        >>> await repo.update(user)
    """
    ...
delete async 
delete(user_id: UUID) -> None

Delete user (soft delete - sets is_active=False).

Parameters:

Name Type Description Default
user_id UUID

User's unique identifier.

 required

Raises:

Type Description
NotFoundError

If user doesn't exist.
DatabaseError

If database operation fails.
Note

This is a soft delete. User remains in database but is_active=False.

Example

await repo.delete(user_id)

Source code in src/domain/protocols/user_repository.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
async def delete(self, user_id: UUID) -> None:
    """Delete user (soft delete - sets is_active=False).

    Args:
        user_id: User's unique identifier.

    Raises:
        NotFoundError: If user doesn't exist.
        DatabaseError: If database operation fails.

    Note:
        This is a soft delete. User remains in database but is_active=False.

    Example:
        >>> await repo.delete(user_id)
    """
    ...
exists_by_email async 
exists_by_email(email: str) -> bool

Check if user with email exists.

Convenience method to check for duplicates before creation.

Parameters:

Name Type Description Default
email str

Email address to check (case-insensitive).

 required

Returns:

Type Description
bool

True if user exists, False otherwise.
Example

if await repo.exists_by_email("user@example.com"): ... raise ConflictError("Email already exists")

Source code in src/domain/protocols/user_repository.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def exists_by_email(self, email: str) -> bool:
    """Check if user with email exists.

    Convenience method to check for duplicates before creation.

    Args:
        email: Email address to check (case-insensitive).

    Returns:
        True if user exists, False otherwise.

    Example:
        >>> if await repo.exists_by_email("user@example.com"):
        ...     raise ConflictError("Email already exists")
    """
    ...
update_password async 
update_password(user_id: UUID, password_hash: str) -> None

Update user's password hash.

Atomic operation that only updates the password field. Used for password reset and password change flows.

Parameters:

Name Type Description Default
user_id UUID

User's unique identifier.

 required
password_hash str

New bcrypt password hash.

 required

Raises:

Type Description
NotFoundError

If user doesn't exist.
DatabaseError

If database operation fails.
Example

await repo.update_password(user_id, new_hash)

Source code in src/domain/protocols/user_repository.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
async def update_password(self, user_id: UUID, password_hash: str) -> None:
    """Update user's password hash.

    Atomic operation that only updates the password field.
    Used for password reset and password change flows.

    Args:
        user_id: User's unique identifier.
        password_hash: New bcrypt password hash.

    Raises:
        NotFoundError: If user doesn't exist.
        DatabaseError: If database operation fails.

    Example:
        >>> await repo.update_password(user_id, new_hash)
    """
    ...
verify_email async 
verify_email(user_id: UUID) -> None

Mark user's email as verified.

Atomic operation that sets is_verified=True. Used for email verification flow.

Parameters:

Name Type Description Default
user_id UUID

User's unique identifier.

 required

Raises:

Type Description
NotFoundError

If user doesn't exist.
DatabaseError

If database operation fails.
Example

await repo.verify_email(user_id)

Source code in src/domain/protocols/user_repository.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
async def verify_email(self, user_id: UUID) -> None:
    """Mark user's email as verified.

    Atomic operation that sets is_verified=True.
    Used for email verification flow.

    Args:
        user_id: User's unique identifier.

    Raises:
        NotFoundError: If user doesn't exist.
        DatabaseError: If database operation fails.

    Example:
        >>> await repo.verify_email(user_id)
    """
    ...