Антипаттерны на питоне, которые меня победили

У нас в компании был один проект, с которым я не справился.

Дело было так: мне сообщили, что знакомый плюсовик написал какой-то скелет проекта на питоне, а теперь мне надо его допилить, подставив в нужные места реализации. Почему плюсовик? Почему тогда делал он, а не я? Почему сейчас должен делать я, а не он? Да чёрт его знает. Но я подумал, что смогу — унаследуюсь и переопределю методы где нужно, подставлю зависимости во всякие DI, ну что там может быть плохого?

Оказалось, что плохо вообще всё.

Проект был раздут и переусложнён настолько, что буквально не умещался у меня в голове. Я уже начинаю подозревать, что, похоже, я LLM, и у меня контекст в районе 4096 токенов, потому что чем дольше я читаю код, тем меньше помню, что там было в начале. Изучая проект, я прыгал по стеку вызовов, пытаясь понять, что откуда вызывается и в какой последовательности, и просто в какой-то момент терялся. Проект меня победил, и его передали кому-то другому.

Недавно мне дали черновик статьи на редактуру, чтобы я её подправил где считаю нужным. А статья-то оказалась как раз про тот самый проект — автор рассказывал, как он классно всё написал. Вы даже не представляете, как сложно писать дифирамбы проекту, который настолько плох и полон антипаттернов, что можно собрать по нему целую статью. Так, подождите-ка, а что если...

Короче, вот она — анти-статья, собранная из того проекта. А где мне не хватало примеров, я брал код из Django, потому что он вообще полностью собран на антипаттернах.

Получилось много букв, как всегда.

? Странная типизация

С типами в питонячьем коде как всегда плохо.

Отсутствие type hints

Когда писали PEP 484, то так и написали: «чтобы kesn не бесился, читая код».

Но все продолжают игнорировать типы. Особенно Django. Народ страдает, пишет стабы, а разрабам Django вообще пофиг, они художники и так видят:

def find_template_loader(self, loader): 	if isinstance(loader, (tuple, list)): 		loader, *args = loader 	else: 		args = []  	if isinstance(loader, str): 		loader_class = import_string(loader) 		return loader_class(self, *args) 	else: 		raise ImproperlyConfigured( 			"Invalid value in template loaders configuration: %r" % loader 		)

Что такое loader? А хрен его знает, но если пройтись по коду, то либо какая-то строка (а точнее, путь к модулю), либо кортеж/список из этой строки и каких-то аргументов, тоже хрен знает каких.

Что тут возвращается? А хрен его знает, мы что-то динамически импортировали (import_string), передали туда себя же (loader_class(self, ...)) и вернули.

Ну типа удачного чтения кода, посоны.

Тут есть два решения — либо писать stubы, либо ездить на конференции, чтобы продвигать тайп хинты в массы встречаться с разработчиками и рассказывать, что вы и есть тот самый склонный к насилию психопат, который читает их код и знает, где они живут.

Неправильные типы

Вообще любой более-менее высокоуровневый язык, во-первых, предоставляет базовый набор типов (иногда очень даже неплохой), а во-вторых, позволяет создавать свои типы и алиасы типов.

Ну вот как в питоне — у нас не только int, float и str, но ещё Decimal, datetime (aware / naive), timedelta, а ещё можно написать так:

type Point = tuple[float, float]

Можно даже создавать distinct тип, то есть в примере ниже Byte он как бы под капотом int, но заменить его на int нельзя (в теории).

from typing import NewType  Byte = NewType('Byte', int)  # define Byte as a distinct type based on int

И это здорово, потому что чем строже у нас типы, тем меньше вероятность по ошибке передать в функцию вместо байтов каких-нибудь крокодилов.

Ещё можно создавать generic-типы:

type ListOrTuple[T] = tuple[T, ...] | list[T]  # Usage list_of_numbers: ListOrTuple[int] = [1, 2, 3] tuple_of_strings: ListOrTuple[str] = "a", "b", "c",

Есть ещё ParamSpec:

from typing import Callable  def logger[**P, R](func: Callable[P, R]) -> Callable[P, R]:     def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:         print("Calling function...")         return func(*args, **kwargs)     return wrapper

И typing.Concatenate (вот только сейчас про него узнал и сразу же пожалел):

from typing import Callable, Concatenate, ParamSpec, TypeVar  P = ParamSpec("P") R = TypeVar("R")  def add_logging[P, R](func: Callable[P, R]) -> Callable[Concatenate[str, P], R]:     def wrapper(log_msg: str, *args: P.args, **kwargs: P.kwargs) -> R:         print(f"Log: {log_msg}")         return func(*args, **kwargs)     return wrapper

Ну то есть типизация в питоне не то что есть, она удовлетворит любое ваше извращённое желание.

Но зачем вся эта сложность существует, если даже в самых тривиальных случаях я постоянно вижу такое:

AWS_OPERATION_RETRY_DELAY_SEC: int = 9

И это ещё повезло, потому что тут автор явно указал _SEC, и из названия понятно, что это секунды. А иногда не указывают, и можно только догадываться, часы там, минуты или секунды. А ведь используй автор нужный тип — и никаких вопросов бы и не было, и вся временна?я арифметика сразу бы работала:

AWS_OPERATION_RETRY_DELAY = timedelta(seconds=9) # теперь можно писать так: retry_datetime = now() + AWS_OPERATION_RETRY_DELAY

Базам данных тоже достаётся, они стараются, изобретают тип INTERVAL, а разработчики в приложении всё равно используют INTEGER:

collect_interval_minutes = models.IntegerField() # а надо было: collect_interval = models.DurationField()

Библиотеки тоже этим грешат, вот тут передаём в timeout не timedelta, а float (количество секунд):

requests.get('https://github.com/', timeout=0.1)

Да даже в стандартной библиотеке постоянно int или float, а не timedelta:

time.sleep(5)

А вот в Playwright, чтоб мы не расслаблялись, сделали все таймауты тоже int, но в миллисекундах:

context.set_default_timeout(5_000)

Бесит.

Перегрузка функций

В питоне нет перегрузки функций (overloading), чтобы можно было написать так:

def foo(a: int) -> int:     return a * 2      def foo(a: str) -> int: 	return foo(int(a))

Красиво ж! Но перегрузка — это скучно, поэтому решили добавить только «намёк на перегрузку» — вы можете использовать специальный декоратор @overload, чтобы помечать места, где могло бы быть красиво, если бы вы выбрали другой язык программирования.

Ну ладно, чего нет, того нет.

Но кто-то не может смириться и делает так, что можно передавать разные типы — например, в requests можно передать таймаут как float или как (float, float):

def request(method, url, **kwargs):	     """     :param timeout: (optional) How many seconds to wait for the server to send data         before giving up, as a float, or a `(connect timeout, read         timeout) <timeouts>` tuple.     :type timeout: float or tuple     """

А в этом методе из Django можно передать строку, а можно кортеж или список. И теперь в коде нужно обрабатывать эти два случая — это легко видно по первому вызову isinstance:

def find_template_loader(self, loader): 	if isinstance(loader, (tuple, list)): 		loader, *args = loader 	else: 		args = []  	if isinstance(loader, str): 		loader_class = import_string(loader) 		return loader_class(self, *args) 	else: 		raise ImproperlyConfigured( 			"Invalid value in template loaders configuration: %r" % loader 		)

Пожалуйста, не надо жертвовать строгостью ради удобства. Да, здорово, что можно написать и так, и так:

find_template_loader("some_loader")  # это всё равно что вызов без arg1 и arg2 find_template_loader(("some_loader", arg1, arg2)) find_template_loader(["some_loader", arg1, arg2])

Но вообще-то ничего страшного не случится, если заставить пользователей пользоваться едиственным вариантом вызова. И внезапно код станет проще:

def find_template_loader(self, loader: str, *args):     loader_class = import_string(loader)     return loader_class(self, *args)

Давайте просто смиримся, что в питоне нет перегрузки.

Dict вместо Class

На интервью любят спрашивать всякую хрень про словари, типа может ли ключом словаря быть кортеж из кабанов, внутри которых фрактал из сухофруктов, и что будет, если фрактал заменить на статую Анубиса?

Да мне плевать, как там можно и что там будет, потому что, по моему мнению, словари — это специальная структура данных, которая связывает пару ключ-значение и, главное, позволяет делать поиск по ключу за константное время. Я так и делаю — использую str или int как ключи и ищу по ним.

Но так как словари все такие гибкие, а питон весь такой динамический, то у разрабов постоянно есть соблазн использовать словарь как универсальную структуру данных. Вот примерчик:

class ShieldTestSettings(ShieldSettings):     model_config = {         'env_file': '.env.test',         'extra': 'ignore',     }

Тут для model_config использовали словарь, но возникают вопросы:

• Что вообще можно передать в model_config? Только env_file и extra, или что-то ещё?

• Какие значения можно использовать для, скажем, extra? ignore и... raise? disallow? forbid?

• Есть ли значения по умолчанию? Что будет, если я не укажу extra?

• Если я случайно опечатаюсь и напишу exxxtra, то что будет? Оно упадёт или проигнорирует?

То есть да, словарь формально хранит данные, но только это он и делает, и он не содержит вообще никакой информации об этих данных и их ограничениях. И это плохо.

Хорошо в данном случае задать явно, что такое model_config, какой у него тип и что он умеет:

@dataclass(slots=True) class ModelConfig: 	env_file: str 	extra: Literal["ignore", "forbid"] = "ignore"   class ShieldTestSettings(ShieldSettings):     model_config = ModelConfig(         env_file='env.test',         extra='ignore',     )

Конечно, существуют случаи, когда словари вполне нормально использовать для временного хранения данных — например, когда нам приходит что-то по API:

response = session.get(API_URL) data: dict = response.json()

Тут нам пришли какие-то пока что неопознанные данные, и мы их записали в data. Приемлемо. Но следующим шагом мы всё равно должны эти данные поместить в какую-то структуру, которая будет их проверять и описывать — например, в pydantic-модель:

user_info = UserInfo.model_validate(data)

Ну то есть словари — это просто какие-то данные в свободном формате без какой-либо дополнительной информации, по которым можно искать по ключу. Понятное дело, что количество случаев, когда нужна именно такая структура данных, не так уж и велико.

? Усложнение кода

Переприсваивание переменной

Создание новой переменой всегда лучше, чем переприсваивание. Следите за ret:

def _validate_manager_state(self) -> bool: 	ret: bool = self._handle_shielded_server_change()  # <-- ret появляется здесь 	# ... 	ret = self._handle_hosted_zone_change() or ret  # <-- и заменяется здесь 	# ... 	return ret  # <-- что возвращается в итоге? непонятно

Тут чтобы понять, что в ret записалось, нужно посмотреть две строчки и что в них происходит. А были бы отдельные переменные...

def _validate_manager_state(self) -> bool: 	shielded_server_changed = self._handle_shielded_server_change() 	# ... 	hosted_zone_changed = self._handle_hosted_zone_change() 	# ... 	return shielded_server_changed or hosted_zone_changed

Видите? Из-за того, что я ещё правильно их назвал, нам даже не нужно читать код, чтобы понять, что возвращается.

Кроме того, из-за того, что ушёл or ret, мы убрали связь между строчками, которые как раз и не должны зависеть друг от друга.

Слишком простой пример? Вот побольше:

def clean_all(self) -> None: 	objects = self.state_manager.get_state().address_manager_created_objects  	cleaned: bool = True 	cleaned = self._clean_objects(objects, WAF, _remove_firewall) and cleaned 	cleaned = self._clean_objects(objects, ELB, _remove_elb) and cleaned 	cleaned = self._clean_objects(objects, SECURITY_GROUP, _remove_security_group) and cleaned 	cleaned = self._clean_objects(objects, TARGET_GROUP, _remove_target_group) and cleaned 	cleaned = self._clean_objects(objects, SUBNET, _remove_subnet) and cleaned 	cleaned = self._clean_objects(objects, VPC, _remove_vpc) and cleaned 	if not cleaned: 		raise AddressManagerException

Тут каждая строчка зависит от предыдущей. Не дай бог вы забудете где-то дописать and cleaned — код развалится. Как бы это переписать? Ну, например, сделать их независимыми друг от друга (убрать and cleaned). А ещё нам не очень-то важно назвать каждый результат, а важно смотреть на результаты в совокупности, так что можно и не создавать переменные, а сложить результаты в массив:

def clean_all(self) -> None: 	objects = self.state_manager.get_state().address_manager_created_objects  	cleaned = [ 		self._clean_objects(objects, WAF, _remove_firewall), 	    self._clean_objects(objects, ELB, _remove_elb), 		self._clean_objects(objects, SECURITY_GROUP, _remove_security_group), 	    self._clean_objects(objects, TARGET_GROUP, _remove_target_group), 	    self._clean_objects(objects, SUBNET, _remove_subnet),         self._clean_objects(objects, VPC, _remove_vpc),     ] 	if not all(cleaned): 		raise AddressManagerException

Есть ещё пример из любви всей моей жизни — Django. Тут loaders — это список строк, а когда я к этому факту привык, то loaders превращается в список, который содержит единственный кортеж вида (строка, список строк):

if loaders is None: 	loaders = ["django.template.loaders.filesystem.Loader"] 	if app_dirs: 		loaders += ["django.template.loaders.app_directories.Loader"] 	loaders = [("django.template.loaders.cached.Loader", loaders)]

Такое я люблю, потому что пока такой код есть, мой мозг напрягается, и деменция мне не грозит. Ну, я надеюсь.

Side effects

Самые приятные на свете функции — это которые получают какие-то данные на вход, делают какую-то магическую обработку внутри себя и что-то возвращают. Вход — вычисления — результат. Изи.

Такие функции хороши как чёрный ящик: даже если внутри много кода, вам это в принципе может быть не важно, пока функция работает как задумано. По аналогии с машиной: вам в принципе не надо знать, как устроена машина, пока она едет.

А теперь давайте посмотрим на какой-нибудь прекрасный метод в Джанго. Например, ChangeList.get_queryset(request, exclude_parameters=None). Вот по названию get_что-то_там кажется, что он безобидный, да? Получает на вход парочку параметров и возвращает что-то. Да, но не только! Ещё этот метод меняет self.filter_specs, self.has_filters, self.has_active_filters и self.clear_all_filters_qs:

def get_queryset(self, request, exclude_parameters=None): 	( 		self.filter_specs,  # <------- ХОБА! 		self.has_filters,  # <------- ХОБА! 		remaining_lookup_params, 		filters_may_have_duplicates, 		self.has_active_filters,  # <------- ХОБА! 	) = self.get_filters(request) 	 	# ...  	self.clear_all_filters_qs = self.get_query_string(  # <------- ХОБА! 		new_params=remaining_lookup_params, 		remove=self.get_filters_params(), 	) 	 	# ... 	 	return qs

Получается, если я в какой-то момент захочу получить значение clear_all_filters_qs, то мне можно сделать это только после вызова get_queryset, потому что ранее это значение ещё не задано.

Та же тема и с классом Form: поле cleaned_data становится доступным как побочный эффект вызова Form.is_valid().

И в этом весь Джанго, там в какой-то момент данных нет, в какой-то момент они есть, и когда эти данные появляются — ни черта не понятно без конкретного вкуривания исходников. Поэтому пишите чистые функции, которые либо меняют какое-то внутренне состояние (self.value = "something"), либо что-то возвращают (return "something"), но не оба этих варианта сразу.

Большие функции

Кошмаром я называю случаи, когда в функциях типа той, что ниже, мне нужно заменить какой-нибудь obj_url. Он вот, на 8 строчке, рукой подать, но чтобы его поменять, мне нужно скопипастить ВЕСЬ этот метод и заменить одну строчку.

def response_add(self, request, obj, post_url_continue=None): 	""" 	Determine the HttpResponse for the add_view stage. 	""" 	opts = obj._meta 	preserved_filters = self.get_preserved_filters(request) 	preserved_qsl = self._get_preserved_qsl(request, preserved_filters) 	obj_url = reverse( 		"admin:%s_%s_change" % (opts.app_label, opts.model_name), 		args=(quote(obj.pk),), 		current_app=self.admin_site.name, 	) 	# Add a link to the object's change form if the user can edit the obj. 	if self.has_change_permission(request, obj): 		obj_repr = format_html('<a href="{}">{}</a>', urlquote(obj_url), obj) 	else: 		obj_repr = str(obj) 	msg_dict = { 		"name": opts.verbose_name, 		"obj": obj_repr, 	} 	# Here, we distinguish between different save types by checking for 	# the presence of keys in request.POST.  	if IS_POPUP_VAR in request.POST: 		to_field = request.POST.get(TO_FIELD_VAR) 		if to_field: 			attr = str(to_field) 		else: 			attr = obj._meta.pk.attname 		value = obj.serializable_value(attr) 		popup_response_data = json.dumps( 			{ 				"value": str(value), 				"obj": str(obj), 			} 		) 		return TemplateResponse( 			request, 			self.popup_response_template 			or [ 				"admin/%s/%s/popup_response.html" 				% (opts.app_label, opts.model_name), 				"admin/%s/popup_response.html" % opts.app_label, 				"admin/popup_response.html", 			], 			{ 				"popup_response_data": popup_response_data, 			}, 		)  	elif "_continue" in request.POST or ( 		# Redirecting after "Save as new". 		"_saveasnew" in request.POST 		and self.save_as_continue 		and self.has_change_permission(request, obj) 	): 		msg = _("The {name} “{obj}” was added successfully.") 		if self.has_change_permission(request, obj): 			msg += " " + _("You may edit it again below.") 		self.message_user(request, format_html(msg, **msg_dict), messages.SUCCESS) 		if post_url_continue is None: 			post_url_continue = obj_url 		post_url_continue = add_preserved_filters( 			{ 				"preserved_filters": preserved_filters, 				"preserved_qsl": preserved_qsl, 				"opts": opts, 			}, 			post_url_continue, 		) 		return HttpResponseRedirect(post_url_continue)  	elif "_addanother" in request.POST: 		msg = format_html( 			_( 				"The {name} “{obj}” was added successfully. You may add another " 				"{name} below." 			), 			**msg_dict, 		) 		self.message_user(request, msg, messages.SUCCESS) 		redirect_url = request.path 		redirect_url = add_preserved_filters( 			{ 				"preserved_filters": preserved_filters, 				"preserved_qsl": preserved_qsl, 				"opts": opts, 			}, 			redirect_url, 		) 		return HttpResponseRedirect(redirect_url)  	else: 		msg = format_html( 			_("The {name} “{obj}” was added successfully."), **msg_dict 		) 		self.message_user(request, msg, messages.SUCCESS) 		return self.response_post_save_add(request, obj)

Вроде норм, но проблемы начнутся, когда я обновлю Джанго. В какой-нибудь версии что-нибудь в оригинальном полотне обязательно изменится, и мне придётся снова копировать всё полотно и заменять в нём obj_url. Я обязательно рано или поздно забуду это сделать, а другой разработчик вообще будет не в курсе, что где-то в кодовой базе у меня лежит пропатченный метод Джанги и за ним надо следить.

Если функция выше кажется вам большой, то что вы скажете насчёт litellm.completion? 3222, мать их за ногу, строки! И самое страшное, что где-то на нашей планете ходят люди, которые это написали и выложили как библиотеку, ещё и приписав в начале:

Thank you ! We ?? you! — Krrish & Ishaan

Спасибо и вам, Криш и Ишан! Кстати, вас не затруднит не трогать клавиатуру годик-другой?

init с миллионом параметров

Это вообще распространённая боль в питоне — повторять одни и те же вещи в __init__. В данном коде каждое из 4 названий (subtensor, netuid, wallet, event_processor) повторили по 4 раза:

class BittensorBlockchainManager(AbstractBlockchainManager):     """     Bittensor BlockchainManager implementation using commitments of knowledge as storage.     """      subtensor: bittensor.Subtensor     netuid: int     wallet: bittensor_wallet.Wallet     event_processor: AbstractMinerShieldEventProcessor      def __init__(         self,         subtensor: bittensor.Subtensor,         netuid: int,         wallet: bittensor_wallet.Wallet,         event_processor: AbstractMinerShieldEventProcessor,     ):         self.subtensor = subtensor         self.netuid = netuid         self.wallet = wallet         self.event_processor = event_processor

Тут вот похуже:

class ShieldClient:     blockchain_manager: AbstractBlockchainManager     certificate: Certificate     encryption_manager: AbstractEncryptionManager     certificate_manager: EDDSACertificateManager     event_processor: AbstractMinerShieldEventProcessor     manifest_manager: ReadOnlyManifestManager     options: ShieldMetagraphOptions     wallet: bittensor_wallet.Wallet      def __init__(         self,         netuid: int,         wallet: bittensor_wallet.Wallet,         event_processor: AbstractMinerShieldEventProcessor,         blockchain_manager: AbstractBlockchainManager,         encryption_manager: AbstractEncryptionManager | None = None,         certificate_manager: EDDSACertificateManager | None = None,         manifest_manager: ReadOnlyManifestManager | None = None,         options: ShieldMetagraphOptions | None = None,     ):         self.netuid = netuid         self.wallet = wallet         self.options = options or ShieldMetagraphOptions()         self.event_processor = event_processor         self.blockchain_manager = blockchain_manager         self.certificate_manager = certificate_manager or self.create_default_certificate_manager()         self.encryption_manager = encryption_manager or self.create_default_encryption_manager()         self.manifest_manager = manifest_manager or self.create_default_manifest_manager(             self.event_processor,             self.encryption_manager,         )

И нет этому конца и края, хотя давно придумали датаклассы, которые решают эту проблему (и некоторые другие, кстати). Смотрите, как уменьшился код:

@dataclass class ShieldClient: 	netuid: int     wallet: bittensor_wallet.Wallet,     options: ShieldMetagraphOptions = field(default_factory=ShieldMetagraphOptions)     manifest_manager: ReadOnlyManifestManager | None = None     event_processor: AbstractMinerShieldEventProcessor     blockchain_manager: AbstractBlockchainManager     encryption_manager: AbstractEncryptionManager | None = None     certificate_manager: EDDSACertificateManager | None = None          def __post_init__(self): 	    self.manifest_manager = self.manifest_manager or self.create_default_manifest_manager(             self.event_processor,             self.encryption_manager,         )         self.certificate_manager = certificate_manager or self.create_default_certificate_manager()         self.encryption_manager = encryption_manager or self.create_default_encryption_manager()

Рекомендую.

Большое количество аргументов

Даёшь больше аргументов богу аргументов!

class MinerShield: 	def __init__( 		self,  		miner_hotkey: Hotkey,  		validators_manager: AbstractValidatorsManager, 		address_manager: AbstractAddressManager,  		manifest_manager: AbstractManifestManager, 		blockchain_manager: AbstractBlockchainManager,  		state_manager: AbstractMinerShieldStateManager, 		event_processor: AbstractMinerShieldEventProcessor, 		options: MinerShieldOptions):

Тут, кстати, разраб почуял, что как-то очень много параметров, и решил часть параметров спрятать в отдельный аргумент options: MinerShieldOptions. Ну типа 11 параметров — это как-то много, а так получается 7 + 1 (в котором 4), вроде норм.

Вообще решение простое — сгруппируйте аргументы:

class ManagersOptions: 	validators_manager: AbstractValidatorsManager 	address_manager: AbstractAddressManager 	manifest_manager: AbstractManifestManager 	blockchain_manager: AbstractBlockchainManager 	state_manager: AbstractMinerShieldStateManager   class MinerShield: 	def __init__( 		self,  		miner_hotkey: Hotkey,  		managers: ManagersOptions, 		event_processor: AbstractMinerShieldEventProcessor,  		options: MinerShieldOptions, 	):

Где эта грань — когда параметров слишком много — хрен знает, вон в litellm только после 40го аргумента подумали: «хм, чёто много получилось, давайте что ли остальное в kwargs засунем»:

def completion(  # type: ignore # noqa: PLR0915     model: str,     # Optional OpenAI params: see https://platform.openai.com/docs/api-reference/chat/create     messages: List = [],     timeout: Optional[Union[float, str, httpx.Timeout]] = None,     temperature: Optional[float] = None,     top_p: Optional[float] = None,     n: Optional[int] = None,     stream: Optional[bool] = None,     stream_options: Optional[dict] = None,     stop=None,     max_completion_tokens: Optional[int] = None,     max_tokens: Optional[int] = None,     modalities: Optional[List[ChatCompletionModality]] = None,     prediction: Optional[ChatCompletionPredictionContentParam] = None,     audio: Optional[ChatCompletionAudioParam] = None,     presence_penalty: Optional[float] = None,     frequency_penalty: Optional[float] = None,     logit_bias: Optional[dict] = None,     user: Optional[str] = None,     # openai v1.0+ new params     reasoning_effort: Optional[         Literal["none", "minimal", "low", "medium", "high", "default"]     ] = None,     verbosity: Optional[Literal["low", "medium", "high"]] = None,     response_format: Optional[Union[dict, Type[BaseModel]]] = None,     seed: Optional[int] = None,     tools: Optional[List] = None,     tool_choice: Optional[Union[str, dict]] = None,     logprobs: Optional[bool] = None,     top_logprobs: Optional[int] = None,     parallel_tool_calls: Optional[bool] = None,     web_search_options: Optional[OpenAIWebSearchOptions] = None,     deployment_id=None,     extra_headers: Optional[dict] = None,     safety_identifier: Optional[str] = None,     service_tier: Optional[str] = None,     # soon to be deprecated params by OpenAI     functions: Optional[List] = None,     function_call: Optional[str] = None,     # set api_base, api_version, api_key     base_url: Optional[str] = None,     api_version: Optional[str] = None,     api_key: Optional[str] = None,     model_list: Optional[list] = None,  # pass in a list of api_base,keys, etc.     # Optional liteLLM function params     thinking: Optional[AnthropicThinkingParam] = None,     # Session management     shared_session: Optional["ClientSession"] = None,     **kwargs, ) -> Union[ModelResponse, CustomStreamWrapper]: 

Хотя по мне, если функция принимает слишком много аргументов, то это признак плохого дизайна. Возможно, функцию стоить разбить на более мелкие.

? Интерфейс + единственная реализация

Вообще интерфейс — хорошая штука. В контексте питона я под интерфейсом подразумеваю абстрактные базовые классы.

Вот вы пишете приложение для учёта кошечек в приюте. Нужно работать с кошечками. Вы пишете класс Cat, у него есть какие-то свойства и методы. Кошечки живут в домиках, и вы добавляете класс House. Вы их используете по всей кодовой базе и всё ХОРОШО.

Но потом возникает мысль: а вдруг завтра нам принесут собачку или трицераптора? А они живут в вольерах или там в пещерах. Значит, теперь нам класса Cat недостаточно, нужен интерфейс Animal, а уже кошечку унаследовать от него. Ну и House не проходит, нужно Habitat, а от него уже унаследоваться.

И нас может быть много таких мест, где нам кажется, что когда-то в будущем нам нужно будет заменить реализацию на другую. С точки зрения архитектуры и расширяемости — классно. С точки зрения отладки и изучения кода — адок. Теперь у нас в два раза больше кода (потому что не просто класс, а интерфейс+класс), и везде вместо конкретной реализации указан интерфейс. Написано animal.do_sound(), а что там на самом деле выполняется в рантайме — я не знаю, потому что do_sound выглядит так:

@abstract_method def do_sound():     ...

И вы мне скажете, что на то интерфейсы и существуют — я не должен смотреть, что там в реализации. А я отвечу, что должен, потому что читать код и отлаживать-то как-то надо! Чем больше у меня потенциальных точек DI, тем сложнее по этому коду перемещаться.

В этом проекте абстрактным было всё! И ладно было бы несколько реализаций на выбор — нет, всегда ровно один интерфейс и рядом одна реализация.

class AbstractEncryptionManager(Generic[PrivateKeyType, PublicKeyType], ABC):     """     Abstract base class for manager handling manifest file encryption.     """      @abstractmethod     def encrypt(self, public_key: PublicKeyType, data: bytes) -> bytes:         """         Encrypts given data using the provided public key. Throws EncryptionError if encryption fails.         """         pass      @abstractmethod     def decrypt(self, private_key: PrivateKeyType, data: bytes) -> bytes:         """         Decrypts given data using the provided private key. Throws DecryptionError if decryption fails.         """         pass   class ECIESEncryptionManager(AbstractEncryptionManager[PrivateKey, PublicKey]):     """     Encryption manager implementation using ECIES algorithm.      Public and private keys are ed25519 keys in hex format.     """      _CURVE: Literal['ed25519'] = 'ed25519'     _ECIES_CONFIG = Config(elliptic_curve=_CURVE)      def encrypt(self, public_key: PublicKey, data: bytes) -> bytes:         try:             return ecies.encrypt(public_key, data, config=self._ECIES_CONFIG)         except Exception as e:             raise EncryptionError(f'Encryption failed: {e}') from e      def decrypt(self, private_key: PrivateKey, data: bytes) -> bytes:         try:             return ecies.decrypt(private_key, data, config=self._ECIES_CONFIG)         except Exception as e:             raise DecryptionError(f'Decryption failed: {e}') from e

Спасибо, но можно не надо? Позвольте я напомню, что если не будет интерфейса, то сам класс реализации уже будет как интерфейс, и его можно будет подменить. В контексте данного примера мне больше нравится так:

class EncryptionManager:     """     Encryption manager implementation using ECIES algorithm.      Public and private keys are ed25519 keys in hex format.     """      _CURVE: Literal['ed25519'] = 'ed25519'     _ECIES_CONFIG = Config(elliptic_curve=_CURVE)      def encrypt(self, public_key: PublicKey, data: bytes) -> bytes:         try:             return ecies.encrypt(public_key, data, config=self._ECIES_CONFIG)         except Exception as e:             raise EncryptionError(f'Encryption failed: {e}') from e      def decrypt(self, private_key: PrivateKey, data: bytes) -> bytes:         try:             return ecies.decrypt(private_key, data, config=self._ECIES_CONFIG)         except Exception as e:             raise DecryptionError(f'Decryption failed: {e}') from e

Это одновременно и интерфейс, и реализация по умолчанию! В большинстве случаев дальше этой самой "реализации по умолчанию" никто не пойдёт (YAGNI), зато 1) у меня кода в 2 раза меньше, и 2) когда я перемещаюсь по коду, я попадаю прямо сразу на реализацию, а не на безликий интерфейс.

Разумеется, в случае, когда реализаций больше одной (ну, например, мы сразу пишем ECIESEncryptionManager и SHAEncryptionManager и даём пользователям выбирать, что им больше нравится), без базового класса / интерфейса никак.

Класс вместо функции

Есть некоторые языки (кхм-кхм), где как-то принято всё делать классами. Иногда эту практику переносят в питон и используют классы там, где хватило бы обычных функций. Получается раздувание и усложнение кода на ровном месте.

Вот тут у нас какой-то обработчик событий. Абстрактный базовый класс (ну разумеется!) с одним методом event, который вызывает другой метод _add_event. А ниже реализация с этим единственным методом _add_event.

class AbstractMinerShieldEventProcessor(ABC):     """     Abstract base class for processor handling events generated by shield.     """      def event(self, template: str, exception: Exception | None = None, **kwargs):         return self._add_event(MinerShieldEvent(template, exception, **kwargs))      @abstractmethod     def _add_event(self, event: MinerShieldEvent):         pass   class PrintingMinerShieldEventProcessor(AbstractMinerShieldEventProcessor):     """     Event processor which logs events to console.     """      def _add_event(self, event: MinerShieldEvent):         print(f'{event.date}: MinerShieldEvent: {event.description} Metadata: {event.metadata}')         if event.exception is not None:             print('Exception happened:')             traceback.print_exception(event.exception, file=stdout)                           manager = ReadOnlyManifestManager(event_processor=PrintingMinerShieldEventProcessor())

Почему, мистер Андерсон, почему вы упорствуете? Почему бы не написать просто функцию и передавать её куда хочется?

def print_event_to_console(template: str, exception: Exception | None = None, **kwargs) -> None: 	event = MinerShieldEvent(template, exception, **kwargs) 	print(f'{event.date}: MinerShieldEvent: {event.description} Metadata: {event.metadata}') 	if event.exception is not None: 		print('Exception happened:') 		traceback.print_exception(event.exception, file=stdout) 		  manager = ReadOnlyManifestManager(event_processor=print_event_to_console)

Правило-то простое: сначала пытаемся решить задачу функцией, а если её не хватает (она становится очень сложной или, например, нужно хранить какое-то внутреннее состояние) - используем класс.

Менеджеры

Как я уже говорил: весь проект, из-за которого появилась эта статья — расширяемый. Нет, не так. РАСШИРЯЕМЫЙ. Поэтому везде должны быть абстрактные базовые классы и менеджеры.

class MinerShield:     """     Main class to be used by Miner to shield himself from DDoS. Call enable() to start the shield. No methods in     managers should be called directly. All operations are done by worker thread. After starting shield user can     schedule tasks to be executed asynchronously.     """      miner_hotkey: Hotkey  # hotkey of shielded miner     worker_thread: Optional[threading.Thread]  # main thread executing tasks     task_queue: Queue['AbstractMinerShieldTask']  # queue of tasks to be executed     run: bool  # flag meaning if shield is running     finishing: bool  # flag meaning if shield should finish its work     ticker: threading.Event  # ticker to be used in ticker_thread     ticker_thread: Optional[threading.Thread]  # thread used to schedule validate operation cyclically     validators_manager: AbstractValidatorsManager  # used to manage validators and their keys     address_manager: AbstractAddressManager  # used to manage public IP/domain addresses assigned to validators     manifest_manager: AbstractManifestManager  # used to manage publishing manifest file     blockchain_manager: AbstractBlockchainManager  # used to manage blockchain operations     state_manager: AbstractMinerShieldStateManager  # used to manage state of the shield     event_processor: AbstractMinerShieldEventProcessor  # used to handle events generated by the shield     options: MinerShieldOptions  # configuration options for shield

Но это как-то слабенько. Вроде гибко, но как будто недостаточно... А что, если мы сделаем абстрактный базовый менеджер внутри абстрактного базового менеджера?

class AbstractManifestManager(ABC):     """     Abstract base class for manager handling publishing manifest file containing encrypted addresses for validators.     """      address_serializer: AbstractAddressSerializer     manifest_serializer: AbstractManifestSerializer     encryption_manager: AbstractEncryptionManager

Ну красота же! Я прям чувствую, как все питонисты интернета полезли клонировать этот проект и расширять своими имплементациями, потому что он такой расширяемый, грех не добавить что-нибудь!

Короче, извините за банальщину, но я щас напишу про YAGNI.

Знаете, раньше я видел светлое будущее своего кода: им вот будут пользоваться, его будут дополнять, вставлять туда свои доработки, итд итп. Потом вдруг осознал, что скорей всего дорабатывать этот код в будущем будет только один человек — я сам. А ещё более вероятно, что не будет никто, потому что либо оно будет норм работать, либо проект закроется. Так нахрена тогда тратить время и строить дорогу там, где никто никогда не поедет?

Постройте обычную дорогу, а если (не «когда», а «если») её станет не хватать — ну что ж, сделаете её пошире, невелика беда. А то получится вот так:

Комментарии и докстринги

Что тут происходит?

class MinerShield:     def enable(self):         if self.worker_thread is not None:             # already started             return          self.finishing = False         self.run = True         self._add_task(MinerShieldInitializeTask())         self.worker_thread = threading.Thread(target=self._worker_function)         self.worker_thread.start()

Запускается какая-то задача MinerShieldInitializeTask, вот она:

class MinerShieldInitializeTask(AbstractMinerShieldTask):     def run(self, miner_shield: MinerShield):         miner_shield._handle_initialize()

Что в ней происходит? Запускается какой-то handle_initialize, вот он:

def _handle_initialize(self): 	self._handle_validate_state(first_run=True) 	self.ticker.clear() 	self.ticker_thread = threading.Thread(target=self._ticker_function) 	self.ticker_thread.start()

Что в нём происходит? Запускаются какие-то _handle_validate_state и _ticker_function, вот они:

def _handle_validate_state(self, first_run: bool = False): 	""" 	Refresh all data and validate current state of shield. If shield is not yet fully working, it will finish 	startup process. If something is wrong, it will be fixed. 	""" 	self._reload_state(first_run) 	validators_changed: bool = self._reload_validators(first_run) 	if self._validate_addresses(): 		validators_changed = True  	if validators_changed: 		self._add_task(MinerShieldValidatorsChangedTask()) 		# do not validate manifest, because it will be updated in the moment 		return  	if not self._validate_manifest_file(): 		self._add_task(MinerShieldUpdateManifestTask()) 		# do not validate manifest info, because it will be updated in the moment 		return  	# this will validate manifest info and will publish new one only if needed     self._add_task(MinerShieldPublishManifestTask())          def _ticker_function(self): 	while not self.ticker.wait(self.options.validate_interval_sec): 		self._add_task(MinerShieldValidateStateTask())

Что в них происходит? Запускаются какие-то MinerShieldValidatorsChangedTask, MinerShieldUpdateManifestTask, MinerShieldPublishManifestTask, MinerShieldValidateStateTask, вот они:

class MinerShieldValidatorsChangedTask(AbstractMinerShieldTask):     def run(self, miner_shield: MinerShield):         # noinspection PyProtectedMember         miner_shield._handle_validators_change()          class MinerShieldUpdateManifestTask(AbstractMinerShieldTask):     def run(self, miner_shield: MinerShield):         # noinspection PyProtectedMember         miner_shield._handle_update_manifest()          class MinerShieldPublishManifestTask(AbstractMinerShieldTask):     def run(self, miner_shield: MinerShield):         # noinspection PyProtectedMember         miner_shield._handle_publish_manifest()          class MinerShieldValidateStateTask(AbstractMinerShieldTask):     def run(self, miner_shield: MinerShield):         # noinspection PyProtectedMember         miner_shield._handle_validate_state()

Что в них происходит? Вызываются методы _handle_validators_change, _handle_update_manifest, _handle_publish_manifest, _handle_validate_state.

Я себя чувствую моряком, которого штормило и бросало из одного метода в другой, пока, наконец, меня не вынесло в сердце шторма, где тихо и вроде даже понятно, какие методы вызываются (наконец-то!).

Автор кода на подсознательном уровне почуял, что чёрт ногу сломит в этом спагетти, и решил переписать... да нет, конечно, ничо он переписывать не стал — он просто добавил описание, что этот метод делает:

class MinerShield:     def enable(self):         """         Enable shield. It starts worker thread, which will do such steps if run for the first time:         1. Fetch validators keys.         2. Creates addresses for all validators.         3. Save manifest file.         4. Publish link to manifest file to blockchain.         5. Eventually close public access to original IP after some time.          It puts events to event_manager after each finished operation. Current state is managed by state_manager.         If any error occurs it is retried forever until shield is disabled.          When shield is running, user can schedule tasks to be processed by worker.         """         ...

И знаете, это отличный пример!

Если для того, чтобы другие вкурили происходящее, нужно написать докстринг и приправить всё комментариями, то, скорее всего, этот код — сомнительного качества. И наоборот, хороший код можно бегло просмотреть и даже без комментариев и докстрингов сразу по названиям понять, что там делается. Что-нибудь такое:

class MinerShield:     def enable(self): 	    if self.is_running: 		    return  		self.add_task(self.fetch_validators) 		self.add_task(self.create_addresses) 		self.add_task(self.save_manifest) 		self.add_task(self.publish_manifest) 		self.add_task(self.close_public_address) 		         self.is_running = True         self.worker_thread = threading.Thread(target=self._worker_function)         self.worker_thread.start()

Поэтому рекомендую настораживаться, когда вам хочется написать объяснение того, что ваш код делает. Возможно, вы переусложнили.

Разделение функционала

Иногда у функции помимо основной цели может появиться дополнительный функционал. Например, при запросе по сети может произойти ошибка, и нужно попробовать получить данные ещё раз (или несколько раз). Большой соблазн этот функционал вшить прямо в тело, например, вот так:

def get_manifest_urls(self, hotkeys): 	try: 		serialized_urls = self.get_metadata(hotkeys) 	except BlockchainManagerException: 		# Retry once 		time.sleep(3) 		serialized_urls = await self.get_metadata(hotkeys) 	...

Тут мы в теле функции пытаемся сделать retry — вторую попытку получения данных. И это плохо, потому что код раздувается, а функционал и его параметры намертво прибиты гвоздями в коде.

Такие дополнительные штуки лучше выносить в отдельные сущности. Конкретно для retry есть всякие готовые библиотеки, но можно написать и что-то своё и неоптимальное, как в примере ниже:

from functools import wraps  def retry(fn: Callable, n_times: int, wait_for: timedelta, exception_class: Type[Exception]):     @wraps(fn) 	def wrapper(*args, **kwargs): 		for _ in range(n_times): 		    try: 		        return fn(*args, **kwargs) 		    except exception_class as exc: 			    last_exc = exc 		        time.sleep(wait_for.total_seconds()) 		raise last_exc 	 	return wrapper 	 def get_manifest_urls(self, hotkeys): 	serialized_urls = retry( 		self.get_metadata(hotkeys), n_times=1, wait_for=3, exception_class=BlockchainManagerException 	) 	...

Мы вынесли дополнительный функционал в декоратор retry, и теперь 1) код целевой функции стал чище и 2) его стало проще настраивать

Фабрики

Ооо, сейчас мне накидают! В общем, сколько я ни работал работу, фабрики мне практически не пригодились.

Давайте подумаем, когда вместо создания чего-то нам нужно что-то, что создаёт что-то? А когда создавать сложно! Такое у меня встречается только в одном месте: в тестах. Представьте, что для теста нужно создать 100 пользователей, у каждого настроить какие-то там права, подключить какие-то дополнительные сущности в связанных табличках итд итп.

Мы убираем сложность создания чего-то внутрь фабрики и практически никогда туда больше не заглядываем.

В проде всё иначе. В проде мне обычно не нужно создавать объекты пачками, а если нужно, то я хочу, чтобы это происходило максимально тупо насколько это возможно. Чем тупее и явнее код в проде, тем легче мне будет понять, что происходит, и отладить в случае ошибки.

Забавно, я сначала вспомнил про этот паттерн, а потом нашёл его в этом проекте. Извините за адок, кода действительно много, ведь это мега-фабрика, которая собирает объект из мини-фабрик. Прикольно же!

class MinerShieldFactory:     """     Factory class to create proper MinerShield instance basing on set environmental variables.     """      @classmethod     def create_miner_shield(cls, settings: ShieldSettings, validators: Iterable[Hotkey] | None = None) -> MinerShield:         """         Args:             settings: ShieldSettings instance.             validators: Dictionary containing validators hotkeys and their public keys.         """         validators_manager: AbstractValidatorsManager = cls.create_validators_manager(settings, validators)         event_processor: AbstractMinerShieldEventProcessor = cls.create_event_processor()         state_manager: AbstractMinerShieldStateManager = cls.create_state_manager(settings)         aws_client_factory: AWSClientFactory = cls.create_aws_client_factory(settings)         address_manager: AbstractAddressManager = cls.create_address_manager(             settings, aws_client_factory, event_processor, state_manager         )         encryption_manager: AbstractEncryptionManager = cls.create_encryption_manager()         manifest_manager: AbstractManifestManager = cls.create_manifest_manager(             settings, encryption_manager, aws_client_factory, event_processor         )         blockchain_manager: AbstractBlockchainManager = cls.create_blockchain_manager(settings, event_processor)          if settings.options.auto_hide_original_server:             raise MinerShieldException('Autohiding is not implemented yet')          return MinerShield(             validators_manager,             address_manager,             manifest_manager,             blockchain_manager,             state_manager,             event_processor,             settings.options,         )      @classmethod     def create_validators_manager(         cls,         settings: ShieldSettings,         validators: Iterable[Hotkey] | None = None,     ) -> AbstractValidatorsManager:         return BittensorValidatorsManager(             subtensor=settings.subtensor.client,             netuid=settings.netuid,             validators=validators,         )      @classmethod     def create_event_processor(cls) -> AbstractMinerShieldEventProcessor:         return PrintingMinerShieldEventProcessor()      @classmethod     def create_state_manager(cls, settings: ShieldSettings) -> AbstractMinerShieldStateManager:         return SQLAlchemyMinerShieldStateManager(settings.sql_alchemy_db_url)      @classmethod     def create_aws_client_factory(cls, settings: ShieldSettings) -> AWSClientFactory:         return AWSClientFactory(settings.aws_access_key_id, settings.aws_secret_access_key, settings.aws_region_name)      @classmethod     def create_address_manager(         cls,         settings: ShieldSettings,         aws_client_factory: AWSClientFactory | None,         event_processor: AbstractMinerShieldEventProcessor,         state_manager: AbstractMinerShieldStateManager,     ) -> AbstractAddressManager:         if aws_client_factory:             return cls.create_aws_address_manager(settings, aws_client_factory, event_processor, state_manager)         else:             raise MinerShieldException('Cannot create address manager')      @classmethod     def create_aws_address_manager(         cls,         settings: ShieldSettings,         aws_client_factory: AWSClientFactory,         event_processor: AbstractMinerShieldEventProcessor,         state_manager: AbstractMinerShieldStateManager,     ) -> AbstractAddressManager:         server_location: ShieldedServerLocation = cls.load_server_aws_location(settings)         return AwsAddressManager(             aws_client_factory, server_location, settings.aws_route53_hosted_zone_id, event_processor, state_manager         )      @classmethod     def create_encryption_manager(cls) -> AbstractEncryptionManager:         return ECIESEncryptionManager()      @classmethod     def create_manifest_manager(         cls,         settings: ShieldSettings,         encryption_manager: AbstractEncryptionManager,         aws_client_factory: AWSClientFactory,         event_processor: AbstractMinerShieldEventProcessor,     ) -> AbstractManifestManager:         manifest_serializer: AbstractManifestSerializer = JsonManifestSerializer()         return S3ManifestManager(             manifest_serializer, encryption_manager, event_processor, aws_client_factory, settings.aws_s3_bucket_name         )      @classmethod     def create_blockchain_manager(         cls,         settings: ShieldSettings,         event_processor: AbstractMinerShieldEventProcessor,     ) -> AbstractBlockchainManager:         return BittensorBlockchainManager(             netuid=settings.netuid,             subtensor=settings.subtensor.client,             wallet=settings.wallet.instance,             event_processor=event_processor,         ) 

Методы-однострочники

Если вам платят за количество напечатанных символов, то я считаю подобное вполне оправданным:

def ban_validator(self, validator_hotkey: Hotkey): 	""" 	Ban a validator by its hotkey. Task will be executed by worker. It will update manifest file and publish info 	about new file version to blockchain. 	""" 	self._add_task(MinerShieldBanValidatorTask(validator_hotkey))

А если нет, то почему бы, блин, не написать в нужном месте просто нужный метод без всяких прослоек?

self._add_task(MinerShieldBanValidatorTask(validator_hotkey))

Ну ладно, в случае выше ban_validator — это как бы интерфейс, с красивым названием и докстрингом. Сомнительно, но окэй.

Но тут мы просто оборачиваем вызов конструктора в вызов функции:

@classmethod def create_default_manifest_manager( 	cls, 	event_processor: AbstractMinerShieldEventProcessor, 	encryption_manager: AbstractEncryptionManager, ) -> ReadOnlyManifestManager: 	return ReadOnlyManifestManager(JsonManifestSerializer(), encryption_manager, event_processor)

А тут просто добавляем алиасы к методам:

class AwsShieldedServerData(BaseModel): 	# ...      def to_json(self) -> str:         return self.model_dump_json()      @staticmethod     def from_json(json_str: str) -> AwsShieldedServerData:         return AwsShieldedServerData.model_validate_json(json_str)

Зачем это усложнение? Почему? Я не знаю. Функция, которая просто вызывает другую функцию — в большинстве случаев безосновательное усложнение.

? Исключения и ошибки

Возврат boolean (success / failure) или None вместо Exceptions

«Boolean головного мозга» пошёл от тех языков, где не было исключений и было принято возвращать либо 0 (означает, что всё хорошо), либо какой-нибудь число, означающее код ошибки. Такое можно видеть в терминале, где команды возвращают не ноль, если что-то пошло не так.

Но питон более высокоуровневый. Я руководствуюсь таким правилом: если отрицательный результат — это тоже результат, то функция может возвращать True или False.

Например, пользователь может быть авторизован на сайте или не авторизован. Второй вариант абсолютно не значит, что произошло что-то плохое, поэтому можно вполне себе завести функцию, которая возвращает True / False:

def is_authorized(user: User) -> bool:

Или вот например поиск, есть ли у канала в телеграме чат-обсуждение. Если чата нет, то это вполне обычная ситуация, тут никакой ошибки нет — поэтому можем вернуть None как признак отсутствия чата:

def get_discussion_chat(client: TelegramClient, channel: Entity) -> Entity | None:     """ Пытаемся найти обсуждение (linked chat) для канала """

А вот если у меня есть метод remove_target_group(self, target_group_id: str), который что-там удаляет, то вариантов нет: либо он успешно удаляет что надо, либо случился какой-то пипец — и для этого в питоне существуют исключения. Но нет, я вижу такое:

def remove_target_group(self, target_group_id: str) -> bool: 	current_server_data: AwsShieldedServerData | None = self._load_server_data()  	error_code: str = '' 	for _ in range(self.AWS_OPERATION_MAX_RETRIES): 		try: 			self.elb_client.delete_target_group(TargetGroupArn=target_group_id) 			break 		except ClientError as e: 			error_code = e.response['Error']['Code'] 			if error_code == 'ResourceInUse': 				time.sleep(self.AWS_OPERATION_RETRY_DELAY_SEC)  # Wait for target group to be deregistered 			else: 				raise e 	else: 		self.event_processor.event( 			'Failed to remove AWS TargetGroup {id}, error={error_code}', id=target_group_id, error_code=error_code 		) 		return False  	self.event_processor.event('Deleted AWS TargetGroup {id}', id=target_group_id) 	self.state_manager.del_address_manager_created_object(AwsObjectTypes.TARGET_GROUP.value, target_group_id) 	return True

Метод возвращает True, если всё норм, а если что-то пошло не так — то False. Или бросает ClientError. Ну что ж, приходится использовать «метод двойного презерватива», чтобы отлавливать ошибки — читаем код ошибки и одновременно ловим исключение:

try: 	result = self._remove_target_group(id_) 	if not result: 	    # обрабатываем случай, когда ResourceInUse except ClientError:     # обрабатываем ошибку сети

А хотелось бы так:

try: 	self._remove_target_group(id_) except ResourceInUseError:     # обрабатываем случай, когда ResourceInUse except Client Error:     # обрабатываем ошибку сети

Нужно выбрать что-то одно — возвращать True / False или бросать исключение. Но ведь питон и сам бросает исключения при нештатных ситуациях, так что выбора у нас нет. Случилась ошибка — бросайте исключение. Не надо возвращать False.

В особо тяжёлых случаях возвращается признак успеха даже там, где он не нужен в принципе. Тут вот разраб чему-то обрадовался и стал писать оптимистичные функции, которые всегда возвращают True:

def _remove_subnet(self, subnet_id: str) -> bool: 	self.ec2_client.delete_subnet(SubnetId=subnet_id) 	self.event_processor.event('Deleted AWS subnet {id}', id=subnet_id) 	self.state_manager.del_address_manager_created_object(AwsObjectTypes.SUBNET.value, subnet_id) 	return True 	 	 def _remove_vpc(self, vpc_id: str) -> bool: 	self.ec2_client.delete_vpc(VpcId=vpc_id) 	self.event_processor.event('Deleted AWS VPC {id}', id=vpc_id) 	self.state_manager.del_address_manager_created_object(AwsObjectTypes.VPC.value, vpc_id) 	return True 

\_(?)_/?

Замалчивание ошибки / non-fail-fast

Язык PHP прекрасен тем, что целиком построен на философии "держаться до последнего" — то есть мы стараемся замолчать любые ошибки и сделать, чтобы хоть что-то куда-то вывелось (напоминает сегодняшние LLM, да?).

Разумеется, «чтобы хоть что-то куда-то вывелось» — это последнее, что вы хотите от вашей программы. Вы хотите детерминизм. Как будто бы бытует мнение, что хрупкие программы — то есть программы, которые падают от каждого чиха - это плохие программы. Я считаю, что это наоборот классные программы — потому что ошибка ловится сразу или взрывает всё, а не живёт в системе и проникает в какие-то критические участки.

Но всё равно я открываю код и вижу:

def get_num_subscribers(text):     """     Parse number of subscribers (supports 'K', 'M', 'B' suffixes).     """     try:         if 'k' in text:             return int(float(text.replace('k', '')) * 1_000)         if 'm' in text:             return int(float(text.replace('m', '')) * 1_000_000)         if 'b' in text:             return int(float(text.replace('b', '')) * 1_000_000_000)  # 'B' ? миллиарды         return int(text)     except ValueError:         logger.exception(f"Parsing errog")

Тут разработчик как бы говорит: я вот написал код для считывания количества подписчиков, но если он не работает, то это значит Дуров криво выводит значения, а у меня всё норм. Стоит телеграму поменять вывод с 1k на 1,234, и код будет говорить, что «сорян, данных нет».

«Попробуй получить адрес, а если не получилось, то и хрен с ним»:

def get_address(self, hotkey: Hotkey) -> Optional[Address]: 	""" 	Get address from blockchain for given user identified by hotkey or None if not found or not valid. 	""" 	serialized_address: Optional[bytes] = self.get(hotkey) 	if serialized_address is None: 		return None 	try: 		return self.address_serializer.deserialize(serialized_address) 	except AddressDeserializationException: 		return None 

Получается, что мы написали код, который не даёт никаких гарантий, что он работает. Он может сегодня выдавать значения, а завтра None, или вперемежку значения и None.

Окей, в некоторых случаях мы можем хотеть такую логику, вроде «о божечки, код, миленький, завтра дедлайн, отработай хоть как-то, забей на ошибки, мы их потом по логам посмотрим и починим». Ну или «показываем страничку товара клиенту побыстрее, а если не подгрузился какой-то баннер — то чёрт с ним».

Но чем больше такого замалчивания ошибок, тем меньше гарантий, что ваш код вообще работает. Обмажьте всё в try-except — и однажды у вас не отработает ничего.

Поэтому я предпочитаю хрупкий код. Пусть везде будут assert, пусть исключения сыпятся отовсюду, а в match-case всегда будет case _: raise. Пусть там, где вы не успели написать код, будет не # TODO, а raise NotImplementedError.

Пусть лучше оно сломается, чем будет работать некорректно и молчать. А если очень страшно получить исключение в каком-то важном месте — например, на той же страничке товара — то от этого защищают тесты.

Нет сомнений

assert

Знаете этот знаменитый график уверенности в себе? Сначала ты ничего не знаешь, потом ты всезнающий бог, а потом ты знаешь, что ничего не знаешь.

С ошибками та же штука: сначала ты уверен, что написал хрень, потом ты думаешь, что пишешь сразу хорошо, а потом опять думаешь, что по умолчанию пишешь неработающую хрень. Вот я из последних — я пишу код, и пока я не покрою его тестами, я очень сильно сомневаюсь, что он работает без ошибок. А когда покрываю тестами, то очень сильно сомневаюсь, что покрыл все случаи.

Но быть неуверенным в своём коде — это даже хорошо. Лучше ставить под сомнение, а не слепо верить как в примере ниже:

@classmethod def _get_hosted_zone_domain(cls, hosted_zone: HostedZone) -> str: 	return hosted_zone.name[:-1]  # Cut '.' from the end of hosted zone name

Тут автор уверен на 100%, что в конце name стоит точка. Может, она там есть сейчас. А может, что-то где-то поменялось — или во внешнем сервисе, или в библиотеке, или в цепочке вызовов, и теперь там этой точки нет. Но коду пофиг, он в любом случае отрубит один символ с конца.

У меня железное правило: в тех случаях, когда я на 100% в чём-то уверен, я пишу assert на это... и потом всё равно ловлю AssertionError в рантайме. Потому что то, как я вижу и понимаю этот мир, зачастую не совпадает с реальностью или может перестать совпадать со временем. Часто это ответы из внешних API, где я на 100% уверен, что нельзя вернуть http код 200, а при этом в теле — ошибку, ну или что-то такое. Но делают же, и имеют право!

В данном случае у нас есть два варианта:

return hosted_zone.name.removesuffix(".") # или assert hosted_zone.name.endswith(".") return hosted_zone.name[:-1]

Первый вариант просто ничего не сделает, если наша картина мира оказалась неверной. Второй вариант нам об этом сообщит, выкинув исключение, что я считаю лучшим вариантом. Лучше упасть, чем сделать непонятно что или работать непонятно с чем.

more_itertools

Помимо того, что more_itertools удобна для работы с коллекциями (разбитие на чанки, группировка, lookahead итд итп), в ней есть ещё one(iterable, too_short=ValueError, too_long=ValueError) — возвращает единственный элемент коллекции или ругается, если элементов нет или больше одного — то есть проверяет, что после всех обработок остался только один элемент в коллекции. Здорово отрезвляет, когда начинаешь использовать для обработки внешних данных :)

Позволяет заменить такой код

 assert len(created_objects[AwsObjectTypes.ELB.value]) == 1, 'only one ELB should be created'  return self._get_elb_info(next(iter(created_objects[AwsObjectTypes.ELB.value])))

на такой:

return self._get_elb_info(one(created_objects[AwsObjectTypes.ELB.value]))

По аналогии с one есть only (возвращает либо единственное значение, либо default, когда нет значений) и strictly_n (возвращает строго n значений).

pydantic

То же правило касается обработки внешних данных. Вот пришёл json-чик. Можно радостно превратить его в dict и взять что-то по ключу. И оно даже работает. Но мы-то уже опытные, поэтому говорим не «оно работает», а «оно пока что работает на тех данных, что мы видели». Поэтому мы не просто работаем с тем, что ожидаем увидеть, а объявляем свою картину мира при помощи pydantic:

from pydantic import BaseModel  # here is what I expect: class Person(BaseModel):     first_name: str     last_name: str     age: int           external_data = requests.get("http://localhost/api/person/1", timeout=5).json() person = Person.model_validate(external_data)

Когда моя картина мира не совпадёт с реальностью — я об этом узнаю.

А ещё обратите внимание, что на самом деле наша картина мира немного глубже, чем указано в модели Person. Например, мы ожидаем, что длина имени и фамилии в разумных пределах, а возраст от 0 до 150. В Person это не указано, и это, конечно, проблема. Будьте как Lego, где чётко указано: ожидаем юзеров от 4 до 99 лет.

Неправильная работа с исключениями

В питоне принято создавать исключения под разные сценарии. Хорошая практика — заменять low-level исключения (типа KeyError или ValueError) нашим специфичным — например, при попытке десериализации произошла какая-то неведомая ошибка, и мы хотим выкинуть не какой-нибудь безликий ValueError, а красивый AddressDeserializationException.

Иногда разрабы так и делают:

def deserialize(self, serialized_data: bytes) -> Address: 	try: 		... 	except Exception as e: 		raise AddressDeserializationException(f"Failed to deserialize address, error='{e}'")

Тут я вижу такие проблемы:

• Дублировать название оригинального исключения ({e}) — избытчно, потому что питон и так в traceback напишет, что пошло не так.

• Строковое отображение исключения может потерять часть информации. str(KeyError(5)) вернёт просто 5, и информация о типе потеряется.

• Питон думает, что наш код сломался, когда пытался обработать оригинальное исключение, и напишет «During handling of the above exception, another exception occurred», хотя по факту AddressDeserializationException — это прямое следствие оригинального Exception. Хорошим тоном считается явное указание этой связи про помощи raise ... from ..., тогда питон напишет «The above exception was the direct cause of the following exception».

• Ну и ловить Exception считается дурным тоном.

Поэтому я бы заменил на это:

def deserialize(self, serialized_data: bytes) -> Address: 	try: 		... 	except (ValueError, SomeOtherError) as e: 		raise AddressDeserializationException from e

? Плохие решения

Тупые названия

Пожалуйста, не нужно придумывать какие-то domain-specific названия там, где есть привычные термины.

В экосистеме bittensor, вокруг которой был написан проект, есть «аксоны», «нейроны», «синапсы» и прочие штуки:

from bittensor.core.chain_data.axon_info import AxonInfo from bittensor.core.chain_data.neuron_info import NeuronInfo

И вы, наверно, думаете, что это что-то про симуляцию мозга, типа там аксоны соединяют нейроны итд итп.

Вот переводчик с шизофазийного языка на программистский:

• «Нейрон» — это просто Node, то есть участник сети.

• «Аксон» — это просто интерфейс доступа к ноде, слушающий на каком-то порту. Ну назовите его Server или Endpoint или API.

• «Дендрит» — это штука для доступа к аксонам. У нормальных людей это назавается Client.

• «Синапсы» гоняют между нейронами через аксоны и дендриты... Тьфу, мракобесие! А можно было бы просто сказать, что это данные (payload), которые отправляют клиенты на ноды по API.

Почти для всего уже изобрели названия, и когда я вижу, что кто-то придумал свой классный молодёжный domain-specific язык, то понимаю, что, скорее всего, это для придания эпичности и важности проекту, а по факту там как всегда банальщина.

Не надо так.

Есть ещё тема с названиями, из которых ничего не понятно. Вот что тут делает метод event? А чёрт его знает, потому что это существительное, а не глагол!

class AbstractMinerShieldEventProcessor(ABC):     def event(self, template: str, exception: Exception | None = None, **kwargs):         ...

Почему просто не называть вещи по тому, что они делают?

Копипаста

Если где-то видна копипаста — скорее всего, это плохой код. На копипасте замыливается глаз и легче совершить ошибку, плюс она раздувает код — вместо переиспользования мы дублируем код.

class MinerShieldInitializeTask(AbstractMinerShieldTask):     def run(self, miner_shield: MinerShield):         # noinspection PyProtectedMember         miner_shield._handle_initialize()   class MinerShieldDisableTask(AbstractMinerShieldTask):     def run(self, miner_shield: MinerShield):         # noinspection PyProtectedMember         miner_shield._handle_disable()   class MinerShieldValidateStateTask(AbstractMinerShieldTask):     def run(self, miner_shield: MinerShield):         # noinspection PyProtectedMember         miner_shield._handle_validate_state()   class MinerShieldValidatorsChangedTask(AbstractMinerShieldTask):     def run(self, miner_shield: MinerShield):         # noinspection PyProtectedMember         miner_shield._handle_validators_change()   class MinerShieldBanValidatorTask(AbstractMinerShieldTask):     def __init__(self, validator_hotkey: Hotkey):         self.validator_hotkey = validator_hotkey      def run(self, miner_shield: MinerShield):         # noinspection PyProtectedMember         miner_shield._handle_ban_validator(self.validator_hotkey)   class MinerShieldUpdateManifestTask(AbstractMinerShieldTask):     def run(self, miner_shield: MinerShield):         # noinspection PyProtectedMember         miner_shield._handle_update_manifest()   class MinerShieldPublishManifestTask(AbstractMinerShieldTask):     def run(self, miner_shield: MinerShield):         # noinspection PyProtectedMember         miner_shield._handle_publish_manifest()

Парадигмы из других языков

Я понимаю, откуда ноги растут. Люди приходит из C++ или Java в наш свободолюбивый и ХХиВП-ный питон и хотят, чтоб было так же строго, как в их языках. Всякие private/public, const и вот это всё.

И знаете, питон настолько гибкий, что можно даже превращать обычные словари в неизменяемые атрибуты класса, как это сделано тут:

class MinerShieldState:     _known_validators: dict[Hotkey, PublicKey]     _banned_validators: dict[Hotkey, datetime]     _validators_addresses: dict[Hotkey, Address]     _manifest_address: Optional[Address]      def __init__(self, known_validators: dict[Hotkey, PublicKey], banned_validators: dict[Hotkey, datetime],                  validators_addresses: dict[Hotkey, Address], manifest_address: Optional[Address]):         super().__setattr__('_known_validators', known_validators)         super().__setattr__('_banned_validators', banned_validators)         super().__setattr__('_validators_addresses', validators_addresses)         super().__setattr__('_manifest_address', manifest_address)      @property     def known_validators(self) -> MappingProxyType[Hotkey, PublicKey]:         return MappingProxyType(self._known_validators)      @property     def banned_validators(self) -> MappingProxyType[Hotkey, datetime]:         return MappingProxyType(self._banned_validators)      @property     def validators_addresses(self) -> MappingProxyType[Hotkey, Address]:         return MappingProxyType(self._validators_addresses)      def __setattr__(self, key, value):         raise AttributeError("State is immutable")      def __delattr__(self, item):         raise AttributeError("State is immutable") 

Можно даже красивее сделать, если вспомнить про @property.setter. Но зачем? Питон — это про свободу, у нас (почти) нет приватных методов / членов класса, и подразумевается, что человек, пользующийся кодом, имеет доступ ко всему и примерно предствляет, что он делает и зачем. И это хорошо!

Поэтому мне кажется, что не стоит пытаться натянуть джаву или плюсы на питона, это всегда выглядит странно. Питон просто другой.

? Сумбурный вывод

Писать плохой код — плохо, а хороший — хорошо.

Ладно, на самом деле я просто поделился болью, и мне стало легче. Ну и наверно важно говорить о проблемах, с которыми постоянно встречаешься. Надеюсь, что-нибудь из статьи было полезно, или вы хотя бы поугорали вместе со мной.

Без ТГ статья УГ, так что Блог погромиста. Там делюсь всякими мыслями и историями из жизни.

Новости, обзоры продуктов и конкурсы от команды Timeweb.Cloud — в нашем Telegram-канале ?