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
|
from typing import Any, Callable, Dict, Optional, Tuple, Union
import sqlalchemy as sa
from szurubooru.model.base import Base
from szurubooru.model.user import User
def get_resource_info(entity: Base) -> Tuple[Any, Any, Union[str, int]]:
serializers = {
"tag": lambda tag: tag.first_name,
"tag_category": lambda category: category.name,
"comment": lambda comment: comment.comment_id,
"post": lambda post: post.post_id,
"pool": lambda pool: pool.pool_id,
"pool_category": lambda category: category.name,
} # type: Dict[str, Callable[[Base], Any]]
resource_type = entity.__table__.name
assert resource_type in serializers
primary_key = sa.inspection.inspect(entity).identity # type: Any
assert primary_key is not None
assert len(primary_key) == 1
resource_name = serializers[resource_type](entity) # type: Union[str, int]
assert resource_name
resource_pkey = primary_key[0] # type: Any
assert resource_pkey
return (resource_type, resource_pkey, resource_name)
def get_aux_entity(
session: Any,
get_table_info: Callable[[Base], Tuple[Base, Callable[[Base], Any]]],
entity: Base,
user: User,
) -> Optional[Base]:
table, get_column = get_table_info(entity)
return (
session.query(table)
.filter(get_column(table) == get_column(entity))
.filter(table.user_id == user.user_id)
.one_or_none()
)
|
