Skip to content

UserMod

Bases: DTMixin, UuidPK, SQLModel

Source code in fastbase/models/UserMod.py 
 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
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
class UserMod(DTMixin, UuidPK, SQLModel):
    email: str = Field(max_length=190, unique=True)
    username: str = Field(max_length=190, unique=True)
    display: str = Field(max_length=199)
    timezone: str | None = Field(max_length=190, default='+0000')

    role: str = Field(max_length=20, default='user')
    groups: list[str] = Field(sa_column=Column(ARRAY(String)), default=[])
    permissions: list[str] = Field(sa_column=Column(ARRAY(String)), default=[])

    is_verified: bool = Field(default=True)                     # TODO: Optional verification
    is_active: bool = Field(default=True)                       # TODO: Optional activation
    banned_at: datetime | None = Field(sa_column=Column(DateTime(timezone=True), nullable=True, index=True))

    def __repr__(self):
        return modstr(self, 'email')

    @property
    def is_super(self) -> bool:
        return self.role == 'super'

    # TESTME: Untested
    @classmethod
    async def get_by_email(cls, session: AsyncSession, email: str) -> Type[Self]:
        raise NotImplementedError()

    # TESTME: Untested
    @classmethod
    async def exists(cls, session: AsyncSession, email: EmailStr) -> bool:
        """Check if a user exists"""
        stmt = select(cls.id).where(cls.email == email)
        execdata = await session.exec(stmt)
        if _ := execdata.first():
            return True

    async def has(self, data: str) -> bool:
        raise NotImplementedError()

    async def attach_group(self, session: AsyncSession, recipient: Self, name: str,
                           *, caching: Callable[[str, list], None] | None = None,
                           async_callback: Callable[[str, list], Awaitable[None]] | None = None):
        """
        Attach a group to user. Prevents duplicates.
        :param session:     AsyncSession
        :param recipient:   The user to recieve the group
        :param name:        Group name
        :param caching:     Callback for caching data
        :param async_callback:    Async callback for generic use
        :raises PermissionsException:
        :return:            None
        """
        # async with AsyncSession(async_engine) as sess:
        # async with asynccontextmanager(get_session)() as sess:
        #     user = await User.get_by_email(sess, 'admin@gmail.com', skip_cache=True)
        def _attach(name_: str) -> list[str]:
            neg_ = f'-{name_}'
            groups_ = set(recipient.groups)

            if neg_ in groups_:
                groups_.remove(neg_)
            elif name_ not in groups_:
                groups_.add(name_)
            return list(groups_)

        if not await self.has('group.attach'):
            raise PermissionsException()

        if not name.strip():
            return

        if not self.is_super and self.email == recipient.email:
            raise AppException('ILLEGAL_ACTION: You cannot modify your own groups.')

        groups = _attach(name)
        recipient.groups = groups
        await session.commit()

        if caching:
            caching(recipient.email, groups)
        if async_callback:
            await async_callback(recipient.email, groups)

    async def detach_group(self, session: AsyncSession, recipient: Self, name: str,
                           *, caching: Callable[[str, list], None] | None = None,
                           async_callback: Callable[[str, list], Awaitable[None]] | None = None):
        """
        Remove a group from user. If the group is a starter group (can't be removed) then a negation is added.
        :param session:     AsyncSession
        :param recipient:   The user who's group is to be removed
        :param name:        Group name
        :param caching:     Callback for caching data
        :param async_callback:    Async callback for generic use
        :raises PermissionsException:
        :return:            None
        """
        def _detach(name_: str) -> list[str]:
            neg_ = f'-{name_}'
            groups_ = set(recipient.groups)

            if name_ in groups_:
                groups_.remove(name_)
            else:
                groups_.add(neg_)
            return list(groups_)

        if not await self.has('group.detach'):
            raise PermissionsException()

        if not name.strip():
            return

        if not self.is_super and self.email == recipient.email:
            raise AppException('ILLEGAL_ACTION: You cannot modify your own groups.')

        groups = _detach(name)
        recipient.groups = groups
        await session.commit()

        if caching:
            caching(recipient.email, list(groups))
        if async_callback:
            await async_callback(recipient.email, list(groups))

    async def attach_permission(self, session: AsyncSession, recipient: Self, perm: str,
                                *, caching: Callable[[str, list], None] | None = None,
                                async_callback: Callable[[str, list], Awaitable[None]] | None = None):
        """
        Attach a permission to user. Prevents duplicates.
        :param session:     AsyncSession
        :param recipient:   The user to recieve the group
        :param perm:        Permission
        :param caching:     Callback for caching data
        :param async_callback:    Async callback for generic use
        :raises PermissionsException:
        :return:            None
        """
        # async with AsyncSession(async_engine) as sess:
        # async with asynccontextmanager(get_session)() as sess:
        #     user = await User.get_by_email(sess, 'admin@gmail.com', skip_cache=True)
        def _attach(name_: str) -> list[str]:
            neg_ = f'-{name_}'
            perms_ = set(recipient.permissions)

            if neg_ in perms_:
                perms_.remove(neg_)
            elif name_ not in perms_:
                perms_.add(name_)
            return list(perms_)

        if not await self.has('permission.attach'):
            raise PermissionsException()

        if not perm.strip():
            return

        if not self.is_super and self.email == recipient.email:
            raise AppException('ILLEGAL_ACTION: You cannot modify your own permissions.')

        permissions = _attach(perm)
        recipient.permissions = permissions
        await session.commit()

        if caching:
            caching(recipient.email, permissions)
        if async_callback:
            await async_callback(recipient.email, permissions)

    async def detach_permission(self, session: AsyncSession, recipient: Self, perm: str,
                                *, caching: Callable[[str, list], None] | None = None,
                                async_callback: Callable[[str, list], Awaitable[None]] | None = None):
        """
        Remove a permission from user. If the permission is a starter group (can't be removed) then a negation is added.
        :param session:     AsyncSession
        :param recipient:   The user who's group is to be removed
        :param perm:        Permission
        :param caching:     Callback for caching data
        :param async_callback:    Async callback for generic use
        :raises PermissionsException:
        :return:            None
        """
        def _detach(perm_: str) -> list[str]:
            neg_ = f'-{perm_}'
            perms_ = set(recipient.permissions)

            if perm_ in perms_:
                perms_.remove(perm_)
            else:
                perms_.add(neg_)
            return list(perms_)

        if not await self.has('permission.detach'):
            raise PermissionsException()

        if not perm.strip():
            return

        if not self.is_super and self.email == recipient.email:
            raise AppException('ILLEGAL_ACTION: You cannot modify your own permissions.')

        permissions = _detach(perm)
        recipient.permissions = permissions
        await session.commit()

        if caching:
            caching(recipient.email, list(permissions))
        if async_callback:
            await async_callback(recipient.email, list(permissions))

    # TESTME: Untested
    async def ban(self, session: AsyncSession, recipient: Self,
                  *, caching: Callable[[str, datetime], None] | None = None,
                  async_callback: Callable[[str, datetime], Awaitable[None]] | None = None):
        """
        Ban user.
        :param session:     AsyncSession
        :param recipient:   The user to ban
        :param caching:     Callback for caching data
        :param async_callback:    Async callback for generic use
        :raises PermissionsException:
        :return:            None
        """
        if not await self.has('ban.attach'):
            raise PermissionsException()

        if recipient.banned_at is not None or self.email == recipient.email:
            return

        now = arrow.utcnow().datetime
        recipient.banned_at = now
        await session.commit()

        if caching:
            caching(recipient.email, now)
        if async_callback:
            await async_callback(recipient.email, now)

    # TESTME: Untested
    async def unban(self, session: AsyncSession, recipient: Self,
                    *, caching: Callable[[str, None], None] | None = None,
                    async_callback: Callable[[str, None], Awaitable[None]] | None = None):
        """
        Unban user.
        :param session:     AsyncSession
        :param recipient:   The user to unban
        :param caching:     Callback for caching data
        :param async_callback:    Async callback for generic use
        :raises PermissionsException:
        :return:            None
        """
        if not await self.has('ban.detach'):
            raise PermissionsException()

        if recipient.banned_at is None or self.email == recipient.email:
            return

        recipient.banned_at = None
        await session.commit()

        if caching:
            caching(recipient.email, None)
        if async_callback:
            await async_callback(recipient.email, None)

Attributes

banned_at: datetime | None = Field(sa_column=Column(DateTime(timezone=True), nullable=True, index=True)) class-attribute instance-attribute

display: str = Field(max_length=199) class-attribute instance-attribute

email: str = Field(max_length=190, unique=True) class-attribute instance-attribute

groups: list[str] = Field(sa_column=Column(ARRAY(String)), default=[]) class-attribute instance-attribute

is_active: bool = Field(default=True) class-attribute instance-attribute

is_super: bool property

is_verified: bool = Field(default=True) class-attribute instance-attribute

permissions: list[str] = Field(sa_column=Column(ARRAY(String)), default=[]) class-attribute instance-attribute

role: str = Field(max_length=20, default='user') class-attribute instance-attribute

timezone: str | None = Field(max_length=190, default='+0000') class-attribute instance-attribute

username: str = Field(max_length=190, unique=True) class-attribute instance-attribute

Functions

__repr__()

Source code in fastbase/models/UserMod.py 
45
46
def __repr__(self):
    return modstr(self, 'email')

attach_group(session, recipient, name, *, caching=None, async_callback=None) async

Attach a group to user. Prevents duplicates.

PARAMETER DESCRIPTION
session

AsyncSession

TYPE: AsyncSession

recipient

The user to recieve the group

TYPE: Self

name

Group name

TYPE: str

caching

Callback for caching data

TYPE: Callable[[str, list], None] | None DEFAULT: None

async_callback

Async callback for generic use

TYPE: Callable[[str, list], Awaitable[None]] | None DEFAULT: None

RETURNS DESCRIPTION

None
RAISES DESCRIPTION
PermissionsException
Source code in fastbase/models/UserMod.py 
 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
async def attach_group(self, session: AsyncSession, recipient: Self, name: str,
                       *, caching: Callable[[str, list], None] | None = None,
                       async_callback: Callable[[str, list], Awaitable[None]] | None = None):
    """
    Attach a group to user. Prevents duplicates.
    :param session:     AsyncSession
    :param recipient:   The user to recieve the group
    :param name:        Group name
    :param caching:     Callback for caching data
    :param async_callback:    Async callback for generic use
    :raises PermissionsException:
    :return:            None
    """
    # async with AsyncSession(async_engine) as sess:
    # async with asynccontextmanager(get_session)() as sess:
    #     user = await User.get_by_email(sess, 'admin@gmail.com', skip_cache=True)
    def _attach(name_: str) -> list[str]:
        neg_ = f'-{name_}'
        groups_ = set(recipient.groups)

        if neg_ in groups_:
            groups_.remove(neg_)
        elif name_ not in groups_:
            groups_.add(name_)
        return list(groups_)

    if not await self.has('group.attach'):
        raise PermissionsException()

    if not name.strip():
        return

    if not self.is_super and self.email == recipient.email:
        raise AppException('ILLEGAL_ACTION: You cannot modify your own groups.')

    groups = _attach(name)
    recipient.groups = groups
    await session.commit()

    if caching:
        caching(recipient.email, groups)
    if async_callback:
        await async_callback(recipient.email, groups)

attach_permission(session, recipient, perm, *, caching=None, async_callback=None) async

Attach a permission to user. Prevents duplicates.

PARAMETER DESCRIPTION
session

AsyncSession

TYPE: AsyncSession

recipient

The user to recieve the group

TYPE: Self

perm

Permission

TYPE: str

caching

Callback for caching data

TYPE: Callable[[str, list], None] | None DEFAULT: None

async_callback

Async callback for generic use

TYPE: Callable[[str, list], Awaitable[None]] | None DEFAULT: None

RETURNS DESCRIPTION

None
RAISES DESCRIPTION
PermissionsException
Source code in fastbase/models/UserMod.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
async def attach_permission(self, session: AsyncSession, recipient: Self, perm: str,
                            *, caching: Callable[[str, list], None] | None = None,
                            async_callback: Callable[[str, list], Awaitable[None]] | None = None):
    """
    Attach a permission to user. Prevents duplicates.
    :param session:     AsyncSession
    :param recipient:   The user to recieve the group
    :param perm:        Permission
    :param caching:     Callback for caching data
    :param async_callback:    Async callback for generic use
    :raises PermissionsException:
    :return:            None
    """
    # async with AsyncSession(async_engine) as sess:
    # async with asynccontextmanager(get_session)() as sess:
    #     user = await User.get_by_email(sess, 'admin@gmail.com', skip_cache=True)
    def _attach(name_: str) -> list[str]:
        neg_ = f'-{name_}'
        perms_ = set(recipient.permissions)

        if neg_ in perms_:
            perms_.remove(neg_)
        elif name_ not in perms_:
            perms_.add(name_)
        return list(perms_)

    if not await self.has('permission.attach'):
        raise PermissionsException()

    if not perm.strip():
        return

    if not self.is_super and self.email == recipient.email:
        raise AppException('ILLEGAL_ACTION: You cannot modify your own permissions.')

    permissions = _attach(perm)
    recipient.permissions = permissions
    await session.commit()

    if caching:
        caching(recipient.email, permissions)
    if async_callback:
        await async_callback(recipient.email, permissions)

ban(session, recipient, *, caching=None, async_callback=None) async

Ban user.

PARAMETER DESCRIPTION
session

AsyncSession

TYPE: AsyncSession

recipient

The user to ban

TYPE: Self

caching

Callback for caching data

TYPE: Callable[[str, datetime], None] | None DEFAULT: None

async_callback

Async callback for generic use

TYPE: Callable[[str, datetime], Awaitable[None]] | None DEFAULT: None

RETURNS DESCRIPTION

None
RAISES DESCRIPTION
PermissionsException
Source code in fastbase/models/UserMod.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
async def ban(self, session: AsyncSession, recipient: Self,
              *, caching: Callable[[str, datetime], None] | None = None,
              async_callback: Callable[[str, datetime], Awaitable[None]] | None = None):
    """
    Ban user.
    :param session:     AsyncSession
    :param recipient:   The user to ban
    :param caching:     Callback for caching data
    :param async_callback:    Async callback for generic use
    :raises PermissionsException:
    :return:            None
    """
    if not await self.has('ban.attach'):
        raise PermissionsException()

    if recipient.banned_at is not None or self.email == recipient.email:
        return

    now = arrow.utcnow().datetime
    recipient.banned_at = now
    await session.commit()

    if caching:
        caching(recipient.email, now)
    if async_callback:
        await async_callback(recipient.email, now)

detach_group(session, recipient, name, *, caching=None, async_callback=None) async

Remove a group from user. If the group is a starter group (can't be removed) then a negation is added.

PARAMETER DESCRIPTION
session

AsyncSession

TYPE: AsyncSession

recipient

The user who's group is to be removed

TYPE: Self

name

Group name

TYPE: str

caching

Callback for caching data

TYPE: Callable[[str, list], None] | None DEFAULT: None

async_callback

Async callback for generic use

TYPE: Callable[[str, list], Awaitable[None]] | None DEFAULT: None

RETURNS DESCRIPTION

None
RAISES DESCRIPTION
PermissionsException
Source code in fastbase/models/UserMod.py 
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
async def detach_group(self, session: AsyncSession, recipient: Self, name: str,
                       *, caching: Callable[[str, list], None] | None = None,
                       async_callback: Callable[[str, list], Awaitable[None]] | None = None):
    """
    Remove a group from user. If the group is a starter group (can't be removed) then a negation is added.
    :param session:     AsyncSession
    :param recipient:   The user who's group is to be removed
    :param name:        Group name
    :param caching:     Callback for caching data
    :param async_callback:    Async callback for generic use
    :raises PermissionsException:
    :return:            None
    """
    def _detach(name_: str) -> list[str]:
        neg_ = f'-{name_}'
        groups_ = set(recipient.groups)

        if name_ in groups_:
            groups_.remove(name_)
        else:
            groups_.add(neg_)
        return list(groups_)

    if not await self.has('group.detach'):
        raise PermissionsException()

    if not name.strip():
        return

    if not self.is_super and self.email == recipient.email:
        raise AppException('ILLEGAL_ACTION: You cannot modify your own groups.')

    groups = _detach(name)
    recipient.groups = groups
    await session.commit()

    if caching:
        caching(recipient.email, list(groups))
    if async_callback:
        await async_callback(recipient.email, list(groups))

detach_permission(session, recipient, perm, *, caching=None, async_callback=None) async

Remove a permission from user. If the permission is a starter group (can't be removed) then a negation is added.

PARAMETER DESCRIPTION
session

AsyncSession

TYPE: AsyncSession

recipient

The user who's group is to be removed

TYPE: Self

perm

Permission

TYPE: str

caching

Callback for caching data

TYPE: Callable[[str, list], None] | None DEFAULT: None

async_callback

Async callback for generic use

TYPE: Callable[[str, list], Awaitable[None]] | None DEFAULT: None

RETURNS DESCRIPTION

None
RAISES DESCRIPTION
PermissionsException
Source code in fastbase/models/UserMod.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
async def detach_permission(self, session: AsyncSession, recipient: Self, perm: str,
                            *, caching: Callable[[str, list], None] | None = None,
                            async_callback: Callable[[str, list], Awaitable[None]] | None = None):
    """
    Remove a permission from user. If the permission is a starter group (can't be removed) then a negation is added.
    :param session:     AsyncSession
    :param recipient:   The user who's group is to be removed
    :param perm:        Permission
    :param caching:     Callback for caching data
    :param async_callback:    Async callback for generic use
    :raises PermissionsException:
    :return:            None
    """
    def _detach(perm_: str) -> list[str]:
        neg_ = f'-{perm_}'
        perms_ = set(recipient.permissions)

        if perm_ in perms_:
            perms_.remove(perm_)
        else:
            perms_.add(neg_)
        return list(perms_)

    if not await self.has('permission.detach'):
        raise PermissionsException()

    if not perm.strip():
        return

    if not self.is_super and self.email == recipient.email:
        raise AppException('ILLEGAL_ACTION: You cannot modify your own permissions.')

    permissions = _detach(perm)
    recipient.permissions = permissions
    await session.commit()

    if caching:
        caching(recipient.email, list(permissions))
    if async_callback:
        await async_callback(recipient.email, list(permissions))

exists(session, email) async classmethod

Check if a user exists

Source code in fastbase/models/UserMod.py 
58
59
60
61
62
63
64
@classmethod
async def exists(cls, session: AsyncSession, email: EmailStr) -> bool:
    """Check if a user exists"""
    stmt = select(cls.id).where(cls.email == email)
    execdata = await session.exec(stmt)
    if _ := execdata.first():
        return True

get_by_email(session, email) async classmethod

Source code in fastbase/models/UserMod.py 
53
54
55
@classmethod
async def get_by_email(cls, session: AsyncSession, email: str) -> Type[Self]:
    raise NotImplementedError()

has(data) async

Source code in fastbase/models/UserMod.py 
66
67
async def has(self, data: str) -> bool:
    raise NotImplementedError()

unban(session, recipient, *, caching=None, async_callback=None) async

Unban user.

PARAMETER DESCRIPTION
session

AsyncSession

TYPE: AsyncSession

recipient

The user to unban

TYPE: Self

caching

Callback for caching data

TYPE: Callable[[str, None], None] | None DEFAULT: None

async_callback

Async callback for generic use

TYPE: Callable[[str, None], Awaitable[None]] | None DEFAULT: None

RETURNS DESCRIPTION

None
RAISES DESCRIPTION
PermissionsException
Source code in fastbase/models/UserMod.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
async def unban(self, session: AsyncSession, recipient: Self,
                *, caching: Callable[[str, None], None] | None = None,
                async_callback: Callable[[str, None], Awaitable[None]] | None = None):
    """
    Unban user.
    :param session:     AsyncSession
    :param recipient:   The user to unban
    :param caching:     Callback for caching data
    :param async_callback:    Async callback for generic use
    :raises PermissionsException:
    :return:            None
    """
    if not await self.has('ban.detach'):
        raise PermissionsException()

    if recipient.banned_at is None or self.email == recipient.email:
        return

    recipient.banned_at = None
    await session.commit()

    if caching:
        caching(recipient.email, None)
    if async_callback:
        await async_callback(recipient.email, None)