mirror of
https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git
synced 2026-07-05 17:32:10 +02:00
Dupe code cleanup
This commit is contained in:
+2
-43
@@ -1,7 +1,6 @@
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, Literal
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel, Field, model_validator
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.path_utils import normalize_contact_route, normalize_route_override
|
||||
|
||||
@@ -38,20 +37,6 @@ class ContactUpsert(BaseModel):
|
||||
last_contacted: int | None = None
|
||||
first_seen: int | None = None
|
||||
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def _translate_legacy_route_fields(cls, data: Any) -> Any:
|
||||
if not isinstance(data, Mapping):
|
||||
return data
|
||||
translated = dict(data)
|
||||
if "direct_path" not in translated and "last_path" in translated:
|
||||
translated["direct_path"] = translated.get("last_path")
|
||||
if "direct_path_len" not in translated and "last_path_len" in translated:
|
||||
translated["direct_path_len"] = translated.get("last_path_len")
|
||||
if "direct_path_hash_mode" not in translated and "out_path_hash_mode" in translated:
|
||||
translated["direct_path_hash_mode"] = translated.get("out_path_hash_mode")
|
||||
return translated
|
||||
|
||||
@classmethod
|
||||
def from_contact(cls, contact: "Contact", **changes) -> "ContactUpsert":
|
||||
return cls.model_validate(
|
||||
@@ -114,20 +99,6 @@ class Contact(BaseModel):
|
||||
direct_route: ContactRoute | None = None
|
||||
route_override: ContactRoute | None = None
|
||||
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def _translate_legacy_route_fields(cls, data: Any) -> Any:
|
||||
if not isinstance(data, Mapping):
|
||||
return data
|
||||
translated = dict(data)
|
||||
if "direct_path" not in translated and "last_path" in translated:
|
||||
translated["direct_path"] = translated.get("last_path")
|
||||
if "direct_path_len" not in translated and "last_path_len" in translated:
|
||||
translated["direct_path_len"] = translated.get("last_path_len")
|
||||
if "direct_path_hash_mode" not in translated and "out_path_hash_mode" in translated:
|
||||
translated["direct_path_hash_mode"] = translated.get("out_path_hash_mode")
|
||||
return translated
|
||||
|
||||
def model_post_init(self, __context) -> None:
|
||||
direct_path, direct_path_len, direct_path_hash_mode = normalize_contact_route(
|
||||
self.direct_path,
|
||||
@@ -186,18 +157,6 @@ class Contact(BaseModel):
|
||||
def has_route_override(self) -> bool:
|
||||
return self.route_override_len is not None
|
||||
|
||||
@property
|
||||
def last_path(self) -> str | None:
|
||||
return self.direct_path
|
||||
|
||||
@property
|
||||
def last_path_len(self) -> int:
|
||||
return self.direct_path_len
|
||||
|
||||
@property
|
||||
def out_path_hash_mode(self) -> int:
|
||||
return self.direct_path_hash_mode
|
||||
|
||||
def effective_route_tuple(self) -> tuple[str, int, int]:
|
||||
if self.has_route_override():
|
||||
return normalize_contact_route(
|
||||
|
||||
@@ -335,23 +335,6 @@ class ContactRepository:
|
||||
)
|
||||
await db.conn.commit()
|
||||
|
||||
@staticmethod
|
||||
async def update_path(
|
||||
public_key: str,
|
||||
path: str,
|
||||
path_len: int,
|
||||
path_hash_mode: int | None = None,
|
||||
updated_at: int | None = None,
|
||||
) -> None:
|
||||
"""Compatibility shim for legacy callers/tests."""
|
||||
await ContactRepository.update_direct_path(
|
||||
public_key,
|
||||
path,
|
||||
path_len,
|
||||
path_hash_mode,
|
||||
updated_at=updated_at,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
async def set_routing_override(
|
||||
public_key: str,
|
||||
|
||||
Reference in New Issue
Block a user