Skip to content

Users

Users Mixins.

Users

Bases: ClientBase

Handles Users Operations.

Source code in src/eventstore_grpc/client/users.py 
 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
class Users(ClientBase):
    """Handles Users Operations."""

    def create_user(
        self,
        login_name: str,
        password: str,
        full_name: str,
        groups: Optional[List[str]] = None,
        **kwargs,
    ) -> users_pb2.CreateResp:
        """Creates a new user."""
        stub = users_pb2_grpc.UsersStub(self.channel)
        if groups is None:
            groups = []
        result = users.create(
            stub,
            login_name=login_name,
            password=password,
            full_name=full_name,
            groups=groups,
            **kwargs,
        )
        return result

    def update_user(
        self,
        login_name: Optional[str] = None,
        password: Optional[str] = None,
        full_name: Optional[str] = None,
        groups: Optional[List[str]] = None,
        **kwargs,
    ) -> users_pb2.UpdateResp:
        """Updates an existing user information."""
        stub = users_pb2_grpc.UsersStub(self.channel)
        updates: dict[str, str | list[str]] = {}
        if login_name is not None:
            updates["login_name"] = login_name
        if password is not None:
            updates["password"] = password
        if full_name is not None:
            updates["full_name"] = full_name
        if groups is not None:
            updates["groups"] = groups
        if not updates:
            raise ValueError("No updates.")
        result = users.update(stub, **updates, **kwargs)  # type: ignore
        return result

    def delete_user(self, login_name: str, **kwargs) -> users_pb2.DeleteResp:
        """Deletes a user."""
        stub = users_pb2_grpc.UsersStub(self.channel)
        result = users.delete(stub, login_name=login_name, **kwargs)
        return result

    def enable_user(self, login_name: str, **kwargs) -> users_pb2.EnableResp:
        """Enables a user."""
        stub = users_pb2_grpc.UsersStub(self.channel)
        result = users.enable(stub, login_name=login_name, **kwargs)
        return result

    def disable_user(self, login_name: str, **kwargs) -> users_pb2.DisableResp:
        """Disables a user."""
        stub = users_pb2_grpc.UsersStub(self.channel)
        result = users.disable(stub, login_name=login_name, **kwargs)
        return result

    def get_user_details(self, login_name: str, **kwargs) -> users_pb2.DetailsResp:
        """Gets details about a user."""
        stub = users_pb2_grpc.UsersStub(self.channel)
        result = users.details(stub, login_name=login_name, **kwargs)
        return next(result)  # type: ignore

    def change_user_password(
        self, login_name: str, current_password: str, new_password: str, **kwargs
    ) -> users_pb2.ChangePasswordResp:
        """Changes a user's password."""
        stub = users_pb2_grpc.UsersStub(self.channel)
        result = users.change_password(
            stub,
            login_name=login_name,
            current_password=current_password,
            new_password=new_password,
            **kwargs,
        )
        return result

    def reset_user_password(
        self, login_name: str, new_password: str, **kwargs
    ) -> users_pb2.ResetPasswordResp:
        """Resets a user's password."""
        stub = users_pb2_grpc.UsersStub(self.channel)
        result = users.reset_password(
            stub, login_name=login_name, new_password=new_password, **kwargs
        )
        return result

change_user_password(login_name, current_password, new_password, **kwargs)

Changes a user's password.

Source code in src/eventstore_grpc/client/users.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
def change_user_password(
    self, login_name: str, current_password: str, new_password: str, **kwargs
) -> users_pb2.ChangePasswordResp:
    """Changes a user's password."""
    stub = users_pb2_grpc.UsersStub(self.channel)
    result = users.change_password(
        stub,
        login_name=login_name,
        current_password=current_password,
        new_password=new_password,
        **kwargs,
    )
    return result

create_user(login_name, password, full_name, groups=None, **kwargs)

Creates a new user.

Source code in src/eventstore_grpc/client/users.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def create_user(
    self,
    login_name: str,
    password: str,
    full_name: str,
    groups: Optional[List[str]] = None,
    **kwargs,
) -> users_pb2.CreateResp:
    """Creates a new user."""
    stub = users_pb2_grpc.UsersStub(self.channel)
    if groups is None:
        groups = []
    result = users.create(
        stub,
        login_name=login_name,
        password=password,
        full_name=full_name,
        groups=groups,
        **kwargs,
    )
    return result

delete_user(login_name, **kwargs)

Deletes a user.

Source code in src/eventstore_grpc/client/users.py 
61
62
63
64
65
def delete_user(self, login_name: str, **kwargs) -> users_pb2.DeleteResp:
    """Deletes a user."""
    stub = users_pb2_grpc.UsersStub(self.channel)
    result = users.delete(stub, login_name=login_name, **kwargs)
    return result

disable_user(login_name, **kwargs)

Disables a user.

Source code in src/eventstore_grpc/client/users.py 
73
74
75
76
77
def disable_user(self, login_name: str, **kwargs) -> users_pb2.DisableResp:
    """Disables a user."""
    stub = users_pb2_grpc.UsersStub(self.channel)
    result = users.disable(stub, login_name=login_name, **kwargs)
    return result

enable_user(login_name, **kwargs)

Enables a user.

Source code in src/eventstore_grpc/client/users.py 
67
68
69
70
71
def enable_user(self, login_name: str, **kwargs) -> users_pb2.EnableResp:
    """Enables a user."""
    stub = users_pb2_grpc.UsersStub(self.channel)
    result = users.enable(stub, login_name=login_name, **kwargs)
    return result

get_user_details(login_name, **kwargs)

Gets details about a user.

Source code in src/eventstore_grpc/client/users.py 
79
80
81
82
83
def get_user_details(self, login_name: str, **kwargs) -> users_pb2.DetailsResp:
    """Gets details about a user."""
    stub = users_pb2_grpc.UsersStub(self.channel)
    result = users.details(stub, login_name=login_name, **kwargs)
    return next(result)  # type: ignore

reset_user_password(login_name, new_password, **kwargs)

Resets a user's password.

Source code in src/eventstore_grpc/client/users.py 
 99
100
101
102
103
104
105
106
107
def reset_user_password(
    self, login_name: str, new_password: str, **kwargs
) -> users_pb2.ResetPasswordResp:
    """Resets a user's password."""
    stub = users_pb2_grpc.UsersStub(self.channel)
    result = users.reset_password(
        stub, login_name=login_name, new_password=new_password, **kwargs
    )
    return result

update_user(login_name=None, password=None, full_name=None, groups=None, **kwargs)

Updates an existing user information.

Source code in src/eventstore_grpc/client/users.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def update_user(
    self,
    login_name: Optional[str] = None,
    password: Optional[str] = None,
    full_name: Optional[str] = None,
    groups: Optional[List[str]] = None,
    **kwargs,
) -> users_pb2.UpdateResp:
    """Updates an existing user information."""
    stub = users_pb2_grpc.UsersStub(self.channel)
    updates: dict[str, str | list[str]] = {}
    if login_name is not None:
        updates["login_name"] = login_name
    if password is not None:
        updates["password"] = password
    if full_name is not None:
        updates["full_name"] = full_name
    if groups is not None:
        updates["groups"] = groups
    if not updates:
        raise ValueError("No updates.")
    result = users.update(stub, **updates, **kwargs)  # type: ignore
    return result