Skip to content

Role

" Represents a database role with associated permissions and optional inheritance. The Role class encapsulates the concept of a database role, including its name, permissions, optional inheritance from another role, and an optional description.

Source code in pum/role_manager.py 
 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
class Role:
    """ "
    Represents a database role with associated permissions and optional inheritance.
    The Role class encapsulates the concept of a database role, including its name,
    permissions, optional inheritance from another role, and an optional description.
    """

    def __init__(
        self,
        name: str,
        permissions: list[Permission] | list[str],
        *,
        inherit: Optional["Role"] = None,
        description: str | None = None,
    ) -> None:
        """Initialize the Role class.
        Args:
            name: Name of the role.
            permissions: List of permissions associated with the role.
            inherit: Optional role to inherit permissions from.
            description: Optional description of the role.
        """
        self.name = name
        if isinstance(permissions, list) and all(isinstance(p, dict) for p in permissions):
            self._permissions = [Permission(**p) for p in permissions]
        elif isinstance(permissions, list) and all(isinstance(p, Permission) for p in permissions):
            self._permissions = permissions
        else:
            raise TypeError("Permissions must be a list of dictionnaries or Permission instances.")

        if inherit is not None and not isinstance(inherit, Role):
            raise TypeError("Inherit must be a Role instance or None.")
        self.inherit = inherit
        self.description = description

    def permissions(self):
        """
        Returns the list of permissions associated with the role.
        """
        return self._permissions

    def exists(self, connection: psycopg.Connection) -> bool:
        """Check if the role exists in the database.
        Args:
            connection: The database connection to execute the SQL statements.
        Returns:
            bool: True if the role exists, False otherwise.
        """
        SqlContent("SELECT 1 FROM pg_roles WHERE rolname = {name}").execute(
            connection=connection,
            commit=False,
            parameters={"name": psycopg.sql.Literal(self.name)},
        ).fetchone() is not None

    def create(
        self, connection: psycopg.Connection, grant: bool = False, commit: bool = False
    ) -> None:
        """Create the role in the database.
        Args:
            connection: The database connection to execute the SQL statements.
            grant: Whether to grant permissions to the role. Defaults to False.
            commit: Whether to commit the transaction. Defaults to False.
        """
        if self.exists(connection):
            logger.info(f"Role {self.name} already exists, skipping creation.")
        else:
            logger.info(f"Creating role {self.name}.")
            SqlContent(
                "CREATE ROLE {name} NOSUPERUSER INHERIT NOCREATEDB NOCREATEROLE NOREPLICATION"
            ).execute(
                connection=connection,
                commit=False,
                parameters={"name": psycopg.sql.Identifier(self.name)},
            )
            if self.description:
                SqlContent("COMMENT ON ROLE {name} IS {description}").execute(
                    connection=connection,
                    commit=False,
                    parameters={
                        "name": psycopg.sql.Identifier(self.name),
                        "description": psycopg.sql.Literal(self.description),
                    },
                )
            if self.inherit:
                SqlContent("GRANT {inherit} TO {role}").execute(
                    connection=connection,
                    commit=False,
                    parameters={
                        "inherit": psycopg.sql.Identifier(self.inherit.name),
                        "role": psycopg.sql.Identifier(self.name),
                    },
                )
        if grant:
            for permission in self.permissions():
                permission.grant(role=self.name, connection=connection, commit=commit)

        if commit:
            connection.commit()

__init__ 

__init__(name: str, permissions: list[Permission] | list[str], *, inherit: Optional[Role] = None, description: str | None = None) -> None

Initialize the Role class. Args: name: Name of the role. permissions: List of permissions associated with the role. inherit: Optional role to inherit permissions from. description: Optional description of the role.

Source code in pum/role_manager.py 
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
def __init__(
    self,
    name: str,
    permissions: list[Permission] | list[str],
    *,
    inherit: Optional["Role"] = None,
    description: str | None = None,
) -> None:
    """Initialize the Role class.
    Args:
        name: Name of the role.
        permissions: List of permissions associated with the role.
        inherit: Optional role to inherit permissions from.
        description: Optional description of the role.
    """
    self.name = name
    if isinstance(permissions, list) and all(isinstance(p, dict) for p in permissions):
        self._permissions = [Permission(**p) for p in permissions]
    elif isinstance(permissions, list) and all(isinstance(p, Permission) for p in permissions):
        self._permissions = permissions
    else:
        raise TypeError("Permissions must be a list of dictionnaries or Permission instances.")

    if inherit is not None and not isinstance(inherit, Role):
        raise TypeError("Inherit must be a Role instance or None.")
    self.inherit = inherit
    self.description = description

create 

create(connection: Connection, grant: bool = False, commit: bool = False) -> None

Create the role in the database. Args: connection: The database connection to execute the SQL statements. grant: Whether to grant permissions to the role. Defaults to False. commit: Whether to commit the transaction. Defaults to False.

Source code in pum/role_manager.py 
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
def create(
    self, connection: psycopg.Connection, grant: bool = False, commit: bool = False
) -> None:
    """Create the role in the database.
    Args:
        connection: The database connection to execute the SQL statements.
        grant: Whether to grant permissions to the role. Defaults to False.
        commit: Whether to commit the transaction. Defaults to False.
    """
    if self.exists(connection):
        logger.info(f"Role {self.name} already exists, skipping creation.")
    else:
        logger.info(f"Creating role {self.name}.")
        SqlContent(
            "CREATE ROLE {name} NOSUPERUSER INHERIT NOCREATEDB NOCREATEROLE NOREPLICATION"
        ).execute(
            connection=connection,
            commit=False,
            parameters={"name": psycopg.sql.Identifier(self.name)},
        )
        if self.description:
            SqlContent("COMMENT ON ROLE {name} IS {description}").execute(
                connection=connection,
                commit=False,
                parameters={
                    "name": psycopg.sql.Identifier(self.name),
                    "description": psycopg.sql.Literal(self.description),
                },
            )
        if self.inherit:
            SqlContent("GRANT {inherit} TO {role}").execute(
                connection=connection,
                commit=False,
                parameters={
                    "inherit": psycopg.sql.Identifier(self.inherit.name),
                    "role": psycopg.sql.Identifier(self.name),
                },
            )
    if grant:
        for permission in self.permissions():
            permission.grant(role=self.name, connection=connection, commit=commit)

    if commit:
        connection.commit()

exists 

exists(connection: Connection) -> bool

Check if the role exists in the database. Args: connection: The database connection to execute the SQL statements. Returns: bool: True if the role exists, False otherwise.

Source code in pum/role_manager.py 
134
135
136
137
138
139
140
141
142
143
144
145
def exists(self, connection: psycopg.Connection) -> bool:
    """Check if the role exists in the database.
    Args:
        connection: The database connection to execute the SQL statements.
    Returns:
        bool: True if the role exists, False otherwise.
    """
    SqlContent("SELECT 1 FROM pg_roles WHERE rolname = {name}").execute(
        connection=connection,
        commit=False,
        parameters={"name": psycopg.sql.Literal(self.name)},
    ).fetchone() is not None

permissions 

permissions()

Returns the list of permissions associated with the role.

Source code in pum/role_manager.py 
128
129
130
131
132
def permissions(self):
    """
    Returns the list of permissions associated with the role.
    """
    return self._permissions