sqlmesh.core.user

 1import typing as t
 2from enum import Enum
 3
 4from sqlmesh.core.notification_target import BasicSMTPNotificationTarget, NotificationTarget
 5from sqlmesh.utils.pydantic import PydanticModel, ValidationInfo, field_validator
 6
 7
 8class UserRole(str, Enum):
 9    """A role to associate the user with"""
10
11    REQUIRED_APPROVER = "required_approver"
12
13    @property
14    def is_required_approver(self) -> bool:
15        return self == UserRole.REQUIRED_APPROVER
16
17
18class User(PydanticModel):
19    """SQLMesh user information that can be used for notifications"""
20
21    username: str
22    """The name to refer to the user"""
23    github_username: t.Optional[str] = None
24    """The github login username"""
25    slack_username: t.Optional[str] = None
26    """The slack username"""
27    email: t.Optional[str] = None
28    """The email for the user (full address)"""
29    roles: t.List[UserRole] = []
30    """List of roles to associate with the user"""
31    notification_targets: t.List[NotificationTarget] = []
32    """List of notification targets"""
33
34    @property
35    def is_required_approver(self) -> bool:
36        """Indicates if this is a required approver for PR approvals."""
37        return UserRole.REQUIRED_APPROVER in self.roles
38
39    @field_validator("notification_targets")
40    def validate_notification_targets(
41        cls,
42        v: t.List[NotificationTarget],
43        info: ValidationInfo,
44    ) -> t.List[NotificationTarget]:
45        email = info.data["email"]
46        for target in v:
47            if isinstance(target, BasicSMTPNotificationTarget) and target.recipients != {email}:
48                raise ValueError("Recipient emails do not match user email")
49        return v