Edit on GitHub

sqlmesh.core.user

View Source

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

class UserRole(builtins.str, enum.Enum): View Source

16class UserRole(str, Enum):
17    """A role to associate the user with"""
18
19    REQUIRED_APPROVER = "required_approver"
20
21    @property
22    def is_required_approver(self) -> bool:
23        return self == UserRole.REQUIRED_APPROVER

A role to associate the user with

REQUIRED_APPROVER = <UserRole.REQUIRED_APPROVER: 'required_approver'>

Inherited Members

enum.Enum: name; value
builtins.str: encode; replace; split; rsplit; join; capitalize; casefold; title; center; count; expandtabs; find; partition; index; ljust; lower; lstrip; rfind; rindex; rjust; rstrip; rpartition; splitlines; strip; swapcase; translate; upper; startswith; endswith; isascii; islower; isupper; istitle; isspace; isdecimal; isdigit; isnumeric; isalpha; isalnum; isidentifier; isprintable; zfill; format; format_map; maketrans

class User(sqlmesh.utils.pydantic.PydanticModel): View Source

26class User(PydanticModel):
27    """SQLMesh user information that can be used for notifications"""
28
29    username: str
30    """The name to refer to the user"""
31    github_username: t.Optional[str] = None
32    """The github login username"""
33    slack_username: t.Optional[str] = None
34    """The slack username"""
35    email: t.Optional[str] = None
36    """The email for the user (full address)"""
37    roles: t.List[UserRole] = []
38    """List of roles to associate with the user"""
39    notification_targets: t.List[NotificationTarget] = []
40    """List of notification targets"""
41
42    @property
43    def is_required_approver(self) -> bool:
44        """Indicates if this is a required approver for PR approvals."""
45        return UserRole.REQUIRED_APPROVER in self.roles
46
47    @field_validator("notification_targets")
48    @field_validator_v1_args
49    def validate_notification_targets(
50        cls,
51        v: t.List[NotificationTarget],
52        values: t.Dict[str, t.Any],
53    ) -> t.List[NotificationTarget]:
54        email = values["email"]
55        for target in v:
56            if isinstance(target, BasicSMTPNotificationTarget) and target.recipients != {email}:
57                raise ValueError("Recipient emails do not match user email")
58        return v

SQLMesh user information that can be used for notifications

username: str

The name to refer to the user

github_username: Union[str, NoneType]

The github login username

slack_username: Union[str, NoneType]

The slack username

email: Union[str, NoneType]

The email for the user (full address)

roles: List[sqlmesh.core.user.UserRole]

List of roles to associate with the user

notification_targets: List[typing_extensions.Annotated[Union[sqlmesh.core.notification_target.BasicSMTPNotificationTarget, sqlmesh.core.notification_target.ConsoleNotificationTarget, sqlmesh.core.notification_target.SlackApiNotificationTarget, sqlmesh.core.notification_target.SlackWebhookNotificationTarget], FieldInfo(annotation=NoneType, required=True, discriminator='type_')]]

List of notification targets

is_required_approver: bool

Indicates if this is a required approver for PR approvals.

@field_validator('notification_targets')

@field_validator_v1_args

47    @field_validator("notification_targets")
48    @field_validator_v1_args
49    def validate_notification_targets(
50        cls,
51        v: t.List[NotificationTarget],
52        values: t.Dict[str, t.Any],
53    ) -> t.List[NotificationTarget]:
54        email = values["email"]
55        for target in v:
56            if isinstance(target, BasicSMTPNotificationTarget) and target.recipients != {email}:
57                raise ValueError("Recipient emails do not match user email")
58        return v

Inherited Members

pydantic.main.BaseModel: BaseModel; model_extra; model_fields_set; model_construct; model_copy; model_dump; model_dump_json; model_json_schema; model_parametrized_name; model_rebuild; model_validate; model_validate_json; model_validate_strings; parse_file; from_orm; construct; schema; schema_json; validate; update_forward_refs
sqlmesh.utils.pydantic.PydanticModel: dict; json; copy; parse_obj; parse_raw; missing_required_fields; extra_fields; all_fields; all_field_infos; required_fields; model_post_init