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
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
|
"""Handle the management of resources."""
from uuid import UUID, uuid4
from typing import Dict, Sequence, NamedTuple
from gn3.auth import db
from gn3.auth.authentication.users import User
from .checks import authorised_p
from .exceptions import AuthorisationError
from .groups import Group, user_group, is_group_leader, authenticated_user_group
class MissingGroupError(AuthorisationError):
"""Raised for any resource operation without a group."""
class ResourceCategory(NamedTuple):
"""Class representing a resource category."""
resource_category_id: UUID
resource_category_key: str
resource_category_description: str
class Resource(NamedTuple):
"""Class representing a resource."""
group: Group
resource_id: UUID
resource_name: str
resource_category: ResourceCategory
public: bool
@authorised_p(("create-resource",), error_message="Could not create resource")
def create_resource(
conn: db.DbConnection, resource_name: str,
resource_category: ResourceCategory) -> Resource:
"""Create a resource item."""
with db.cursor(conn) as cursor:
group = authenticated_user_group(conn).maybe(False, lambda val: val)
if not group:
raise MissingGroupError(
"User with no group cannot create a resource.")
resource = Resource(group, uuid4(), resource_name, resource_category, False)
cursor.execute(
"INSERT INTO resources VALUES (?, ?, ?, ?, ?)",
(str(resource.group.group_id), str(resource.resource_id),
resource_name,
str(resource.resource_category.resource_category_id),
1 if resource.public else 0))
return resource
def resource_categories(conn: db.DbConnection) -> Sequence[ResourceCategory]:
"""Retrieve all available resource categories"""
with db.cursor(conn) as cursor:
cursor.execute("SELECT * FROM resource_categories")
return tuple(
ResourceCategory(UUID(row[0]), row[1], row[2])
for row in cursor.fetchall())
return tuple()
def public_resources(conn: db.DbConnection) -> Sequence[Resource]:
"""List all resources marked as public"""
categories = {
str(cat.resource_category_id): cat for cat in resource_categories(conn)
}
with db.cursor(conn) as cursor:
cursor.execute("SELECT * FROM resources WHERE public=1")
results = cursor.fetchall()
group_uuids = tuple(row[0] for row in results)
query = ("SELECT * FROM groups WHERE group_id IN "
f"({', '.join(['?'] * len(group_uuids))})")
cursor.execute(query, group_uuids)
groups = {row[0]: Group(UUID(row[0]), row[1]) for row in cursor.fetchall()}
return tuple(
Resource(groups[row[0]], UUID(row[1]), row[2], categories[row[3]],
bool(row[4]))
for row in results)
def group_leader_resources(
cursor: db.DbCursor, user: User, group: Group,
res_categories: Dict[UUID, ResourceCategory]) -> Sequence[Resource]:
"""Return all the resources available to the group leader"""
if is_group_leader(cursor, user, group):
cursor.execute("SELECT * FROM resources WHERE group_id=?",
(str(group.group_id),))
return tuple(
Resource(group, UUID(row[1]), row[2], res_categories[UUID(row[3])],
bool(row[4]))
for row in cursor.fetchall())
return tuple()
def user_resources(conn: db.DbConnection, user: User) -> Sequence[Resource]:
"""List the resources available to the user"""
categories = { # Repeated in `public_resources` function
cat.resource_category_id: cat for cat in resource_categories(conn)
}
with db.cursor(conn) as cursor:
group = user_group(cursor, user).maybe(False, lambda val: val) # type: ignore[misc]
if not group:
return public_resources(conn)
gl_resources = group_leader_resources(cursor, user, group, categories)
cursor.execute(
("SELECT resources.* FROM group_user_roles_on_resources "
"LEFT JOIN resources "
"ON group_user_roles_on_resources.resource_id=resources.resource_id "
"WHERE group_user_roles_on_resources.group_id = ? "
"AND group_user_roles_on_resources.user_id = ?"),
(str(group.group_id), str(user.user_id)))
private_res = tuple(
Resource(group, UUID(row[1]), row[2], categories[UUID(row[3])],
bool(row[4]))
for row in cursor.fetchall())
return tuple(set(private_res).union(gl_resources).union(public_resources(conn)))
|
