# при отправке
await writer.write(b'screen ')
# отправляем длину как 4байтовое беззнаковое целое с порядком бит BigEndian
# можно выбрать эти параметры иными, лишь бы отправитель и получатель делали одинаково.
await writer.write(len(image_data).to_bytes(4, 'big', signed=False))
await writer.write(image_data)
#при приёме
received = await reader.read(2048)
MAGIC = b'screen '
if received.startswith(MAGIC):
received = received[len(MAGIC):] # отрезаем сигнатуру
length_bytes, received = received[:4], received[4:] # вынимаем из буфера байты длины
# декодируем длину как 4байтовое беззнаковое целое с порядком бит BigEndian
image_length = int.from_bytes(length_bytes, 'big', signed=False)
# дочитываем остаток изображения! Вот этого я от тебя так и не увидел.
while (len(received) < image_length):
received += await reader.read(2048)
# на случай если после изображения будут ещё данные
screenshot, received = received[:image_length], received[image_length:]
# в screenshot должно лежать полное изображениеclass APIError(Exception):
pass
class InvalidToken(APIError):
pass
class APIWrapper:
def __init__(self, token=None):
self.token = token # публичное поле
def _acquire_new_token(self):
response = ask_actual_api_for_token() # пусть мы получаем словарь из JSON
if error := response.get('error', None):
raise APIError(error)
else:
self.token = response['token']
@requires_token
def do_stuff(self):
response = ask_actual_api_for_stuff(self.token, params='foobar')
if error := response.get('error', None):
if error == 'Invalid token':
raise InvalidToken()
else:
raise APIError(error)
# делаем что нужно с данными
do_stuff_with_response(response)def hello(x):
print('hello,', x)
print('это выполнится в любом случае')
if __name__ == '__main__':
print('это выполнится, только если скрипт запущен непосредственно')
hello('world!')
else:
print('это выполнится, только если скрипт импортирован в другой')
Вообще никто не запрещает описать преобразование словарём, если нужна универсальность.
Только не надо забывать про принцип KISS.