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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
# SPDX-License-Identifier: GPL-3.0-or-later
# Haketilo proxy data and configuration (PayloadRef subtype).
#
# This file is part of Hydrilla&Haketilo.
#
# Copyright (C) 2022 Wojtek Kosior
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
#
# I, Wojtek Kosior, thereby promise not to sue for violation of this
# file's license. Although I request that you do not make use this code
# in a proprietary program, I am not going to enforce this in court.
"""
This module provides an interface to interact with payloads inside Haketilo.
"""
# Enable using with Python 3.7.
from __future__ import annotations
import dataclasses as dc
import typing as t
from .. import state as st
from . import base
@dc.dataclass(frozen=True, unsafe_hash=True)
class ConcretePayloadRef(st.PayloadRef):
state: base.HaketiloStateWithFields = dc.field(hash=False, compare=False)
def get_data(self) -> st.PayloadData:
try:
return self.state.payloads_data[self]
except KeyError:
raise st.MissingItemError()
def get_mapping(self) -> st.MappingVersionRef:
raise NotImplementedError()
def get_script_paths(self) \
-> t.Iterable[t.Sequence[str]]:
with self.state.cursor() as cursor:
cursor.execute(
'''
SELECT
i.identifier, fu.name
FROM
payloads AS p
LEFT JOIN resolved_depended_resources AS rdd
USING (payload_id)
LEFT JOIN item_versions AS iv
ON rdd.resource_item_id = iv.item_version_id
LEFT JOIN items AS i
USING (item_id)
LEFT JOIN file_uses AS fu
USING (item_version_id)
WHERE
fu.type = 'W' AND
p.payload_id = ? AND
(fu.idx IS NOT NULL OR rdd.idx IS NULL)
ORDER BY
rdd.idx, fu.idx;
''',
(self.id,)
)
paths: list[t.Sequence[str]] = []
for resource_identifier, file_name in cursor.fetchall():
if resource_identifier is None:
# payload found but it had no script files
return ()
paths.append((resource_identifier, *file_name.split('/')))
if paths == []:
# payload not found
raise st.MissingItemError()
return paths
def get_file_data(self, path: t.Sequence[str]) \
-> t.Optional[st.FileData]:
if len(path) == 0:
raise st.MissingItemError()
resource_identifier, *file_name_segments = path
file_name = '/'.join(file_name_segments)
with self.state.cursor() as cursor:
cursor.execute(
'''
SELECT
f.data, fu.mime_type
FROM
payloads AS p
JOIN resolved_depended_resources AS rdd
USING (payload_id)
JOIN item_versions AS iv
ON rdd.resource_item_id = iv.item_version_id
JOIN items AS i
USING (item_id)
JOIN file_uses AS fu
USING (item_version_id)
JOIN files AS f
USING (file_id)
WHERE
p.payload_id = ? AND
i.identifier = ? AND
fu.name = ? AND
fu.type = 'W';
''',
(self.id, resource_identifier, file_name)
)
result = cursor.fetchall()
if result == []:
return None
(data, mime_type), = result
return st.FileData(type=mime_type, name=file_name, contents=data)
|
