models.User

src/idserver/models/User.py
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
from .SqlAlchemyBase import Base
from sqlalchemy import Column, String, Integer, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from .GUID import GUID
from .IdentityProvider import DbIdentityProvider
from .DbUserClaim import DbUserClaim
from typing import Dict, Any, List, Optional
import uuid
from datetime import datetime

class User:

    def __init__(self, provider_name: str,
                    provider_subject: str, 
                    email: str, 
                    given_name: str,
                    family_name: Optional[str],
                    created_on: int=0, 
                    claims: Dict[str, Any]={}, 
                    last_update: Optional[int]=None,
                    subject: Optional[uuid.UUID]=None) -> None:

        self.subject = subject
        self.provider_name = provider_name
        self.provider_subject = provider_subject
        self.email = email
        self.given_name = given_name
        self.family_name = family_name
        self.created_on = created_on
        self.claims = claims
        self.last_update = last_update

class DbUser(Base):
    __tablename__ = 'users'

    subject = Column(GUID(), primary_key=True)
    provider_subject = Column(String(256), unique=True)
    provider_id = Column(Integer, ForeignKey('identity_provider.provider_id'))
    email = Column(String(256))
    given_name = Column(String(128), nullable=True)
    family_name = Column(String(128))
    created_on = Column(DateTime())
    last_update = Column(DateTime(), nullable=True)

    claims: List[DbUserClaim] = relationship('DbUserClaim', lazy="joined")
    provider: DbIdentityProvider = relationship('DbIdentityProvider', lazy="joined")

    def toUser(self) -> User:
        claims: Dict[str, Any] = {}
        for claim in self.claims:
            claims[claim.claim_name] = claim.claim_value

        last_update: Optional[int] = None
        if self.last_update:
            last_update = self.last_update.timestamp()

        return User(subject=self.subject, 
                    provider_subject=self.provider_subject, 
                    provider_name=self.provider.provider_name,
                    email=self.email, 
                    given_name=self.given_name,
                    family_name=self.family_name,
                    created_on=self.created_on.timestamp(),
                    claims=claims,
                    last_update=last_update)