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
|
from typing import Optional, List, Dict
from datetime import datetime
from szurubooru import db, model, search, rest
from szurubooru.func import auth, pools, snapshots, serialization, versions
_search_executor = search.Executor(search.configs.PoolSearchConfig())
def _serialize(ctx: rest.Context, pool: model.Pool) -> rest.Response:
return pools.serialize_pool(
pool, options=serialization.get_serialization_options(ctx))
def _get_pool(params: Dict[str, str]) -> model.Pool:
return pools.get_pool_by_id(params['pool_id'])
@rest.routes.get('/pools/?')
def get_pools(
ctx: rest.Context, _params: Dict[str, str] = {}) -> rest.Response:
auth.verify_privilege(ctx.user, 'pools:list')
return _search_executor.execute_and_serialize(
ctx, lambda pool: _serialize(ctx, pool))
@rest.routes.post('/pool/?')
def create_pool(
ctx: rest.Context, _params: Dict[str, str] = {}) -> rest.Response:
auth.verify_privilege(ctx.user, 'pools:create')
names = ctx.get_param_as_string_list('names')
category = ctx.get_param_as_string('category')
description = ctx.get_param_as_string('description', default='')
posts = ctx.get_param_as_int_list('posts', default=[])
pool = pools.create_pool(names, category, posts)
pool.last_edit_time = datetime.utcnow()
pools.update_pool_description(pool, description)
ctx.session.add(pool)
ctx.session.flush()
snapshots.create(pool, ctx.user)
ctx.session.commit()
return _serialize(ctx, pool)
@rest.routes.get('/pool/(?P<pool_id>[^/]+)/?')
def get_pool(ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
auth.verify_privilege(ctx.user, 'pools:view')
pool = _get_pool(params)
return _serialize(ctx, pool)
@rest.routes.put('/pool/(?P<pool_id>[^/]+)/?')
def update_pool(ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
pool = _get_pool(params)
versions.verify_version(pool, ctx)
versions.bump_version(pool)
if ctx.has_param('names'):
auth.verify_privilege(ctx.user, 'pools:edit:names')
pools.update_pool_names(pool, ctx.get_param_as_string_list('names'))
if ctx.has_param('category'):
auth.verify_privilege(ctx.user, 'pools:edit:category')
pools.update_pool_category_name(
pool, ctx.get_param_as_string('category'))
if ctx.has_param('description'):
auth.verify_privilege(ctx.user, 'pools:edit:description')
pools.update_pool_description(
pool, ctx.get_param_as_string('description'))
if ctx.has_param('posts'):
auth.verify_privilege(ctx.user, 'pools:edit:posts')
posts = ctx.get_param_as_int_list('posts')
pools.update_pool_posts(pool, posts)
pool.last_edit_time = datetime.utcnow()
ctx.session.flush()
snapshots.modify(pool, ctx.user)
ctx.session.commit()
return _serialize(ctx, pool)
@rest.routes.delete('/pool/(?P<pool_id>[^/]+)/?')
def delete_pool(ctx: rest.Context, params: Dict[str, str]) -> rest.Response:
pool = _get_pool(params)
versions.verify_version(pool, ctx)
auth.verify_privilege(ctx.user, 'pools:delete')
snapshots.delete(pool, ctx.user)
pools.delete(pool)
ctx.session.commit()
return {}
@rest.routes.post('/pool-merge/?')
def merge_pools(
ctx: rest.Context, _params: Dict[str, str] = {}) -> rest.Response:
source_pool_id = ctx.get_param_as_string('remove')
target_pool_id = ctx.get_param_as_string('mergeTo')
source_pool = pools.get_pool_by_id(source_pool_id)
target_pool = pools.get_pool_by_id(target_pool_id)
versions.verify_version(source_pool, ctx, 'removeVersion')
versions.verify_version(target_pool, ctx, 'mergeToVersion')
versions.bump_version(target_pool)
auth.verify_privilege(ctx.user, 'pools:merge')
pools.merge_pools(source_pool, target_pool)
snapshots.merge(source_pool, target_pool, ctx.user)
ctx.session.commit()
return _serialize(ctx, target_pool)
|
