chore: улучшен интерфейс работы с моделями

src/App/bot/interface.py

@@ -5,7 +5,7 @@ from urllib.parse import urlsplit
 from Telegram.interfaces import Interface, command, register, state, text
 
 from .. import services, tasks
-from ..models import Point, Server
+from ..models import Credentials, Point, Server
 from . import texts
 
 # состояния диалога
@@ -26,14 +26,14 @@ PROXY_SCHEMES = ('http', 'https', 'socks5')
 HTML = 'HTML'
 
 
-def parse_credentials(value: str) -> services.Credentials | None:
+def parse_credentials(value: str) -> Credentials | None:
 	"""'ip:password' -> Credentials. Пароль может содержать двоеточия."""
 	address, _, password = value.strip().partition(':')
 	try:
 		address = str(ipaddress.ip_address(address))
 	except ValueError:
 		return None
-	return services.Credentials(address, password) if password else None
+	return Credentials(address, password) if password else None
 
 
 def is_domain(value: str) -> bool:
@@ -58,7 +58,7 @@ class MyPointInterface(Interface):
 	@text(texts.MAIN_MENU, texts.CANCEL)
 	def main_menu(self, ctx):
 		ctx.reset()
-		points = list(ctx.user.points.order_by('created_at'))
+		points = list(Point.objects.owned_by(ctx.user))
 		if points:
 			ctx.reply(texts.points_list(points), keyboard=texts.MAIN_KEYBOARD_WITH_POINTS, parse_mode=HTML)
 		else:
@@ -136,7 +136,7 @@ class MyPointInterface(Interface):
 		other = draft.get('mask' if key == 'server' else 'server', {})
 		if other.get('address') == credentials.address:
 			return ctx.reply(texts.ADDRESS_DUPLICATE)
-		if Server.objects.filter(address=credentials.address).exists():
+		if Server.objects.is_address_taken(credentials.address):
 			return ctx.reply(texts.ADDRESS_TAKEN.format(address=credentials.address))
 		draft[key] = {'address': credentials.address, 'password': credentials.password}
 		self.show_draft(ctx)
@@ -195,10 +195,10 @@ class MyPointInterface(Interface):
 			return ctx.reply(texts.MASK_WITHOUT_DOMAIN)
 		if draft.get('cloudflare') and not draft.get('domain'):
 			return ctx.reply(texts.CLOUDFLARE_WITHOUT_DOMAIN)
-		point = services.create_point(
+		point = Point.objects.create_point(
 			ctx.user,
-			panel=services.Credentials(**draft['server']),
-			mask=services.Credentials(**draft['mask']) if draft.get('mask') else None,
+			panel=Credentials(**draft['server']),
+			mask=Credentials(**draft['mask']) if draft.get('mask') else None,
 			domain=draft.get('domain', ''),
 			cloudflare=draft.get('cloudflare', False),
 			outbound_proxy=draft.get('proxy', ''),
@@ -212,12 +212,12 @@ class MyPointInterface(Interface):
 	@text(texts.MANAGE)
 	def ask_point(self, ctx):
 		ctx.reset()
-		points = ctx.user.points.order_by('created_at')
+		points = Point.objects.owned_by(ctx.user)
 		self.ask(ctx, SELECT, f'{texts.SELECT_PROMPT}\n{texts.points_list(points)}')
 
 	@state(SELECT)
 	def select_point(self, ctx):
-		point = ctx.user.points.filter(name=ctx.text.strip()).first()
+		point = Point.objects.owned_by(ctx.user).find_by_name(ctx.text.strip())
 		if point is None:
 			return ctx.reply(texts.POINT_NOT_FOUND)
 		self.show_point(ctx, point)
@@ -225,14 +225,14 @@ class MyPointInterface(Interface):
 	@text(texts.SERVER_MENU)
 	def server_menu(self, ctx):
 		"""Кнопка из уведомления: последняя открытая точка или последняя созданная."""
-		point = self.current_point(ctx) or ctx.user.points.order_by('-created_at').first()
+		point = self.current_point(ctx) or Point.objects.owned_by(ctx.user).find_last()
 		if point is None:
 			return self.main_menu(ctx)
 		self.show_point(ctx, point)
 
 	def current_point(self, ctx) -> Point | None:
 		point_id = ctx.data.get('point')
-		return ctx.user.points.filter(pk=point_id).first() if point_id else None
+		return Point.objects.owned_by(ctx.user).find_by_pk(point_id) if point_id else None
 
 	def show_point(self, ctx, point: Point):
 		ctx.reset()
@@ -260,8 +260,7 @@ class MyPointInterface(Interface):
 		point = self.idle_point(ctx)
 		if point is None:
 			return
-		point.status = Point.Status.RELEASING
-		point.save(update_fields=['status'])
+		point.set_status(Point.Status.RELEASING)
 		tasks.release_point.send(point.pk, ctx.chat.pk)
 		ctx.reset()
 		ctx.reply(texts.RELEASE_STARTED, keyboard=texts.MAIN_KEYBOARD_WITH_POINTS)
@@ -280,7 +279,7 @@ class MyPointInterface(Interface):
 		point = self.idle_point(ctx)
 		if point is None:
 			return
-		if not point.servers.filter(address=credentials.address).exists():
+		if point.servers.find_by_address(credentials.address) is None:
 			return ctx.reply(texts.REINSTALL_UNKNOWN_ADDRESS.format(address=credentials.address))
 		self.start_reinstall(ctx, point, credentials.address, credentials.password)
 
@@ -291,8 +290,7 @@ class MyPointInterface(Interface):
 			self.start_reinstall(ctx, point)
 
 	def start_reinstall(self, ctx, point: Point, address: str | None = None, password: str | None = None):
-		point.status = Point.Status.INSTALLING
-		point.save(update_fields=['status'])
+		point.set_status(Point.Status.INSTALLING)
 		tasks.reinstall_point.send(point.pk, ctx.chat.pk, address, password)
 		ctx.reset()
 		ctx.reply(texts.REINSTALL_STARTED, keyboard=texts.MAIN_KEYBOARD_WITH_POINTS)