erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin identity: Use Pydantic models:

Browse source 
Use Pydantic models for data used with bridge.
Also update `backend.models.core.DiscoIdentity` to have `None` as possible and default
value for `name`.
This commit is contained in:
Goffi 2026-08-12 18:12:38 +02:00
commit 93ce4112f8
2 changed files with 68 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

2
libervia/backend/models/core.py
View file

@ -85,6 +85,6 @@ class DiscoIdentity(BaseModel):
Note that an XMPP entity may have several disco identities.
"""
name: str
name: str | None = None
category: str
type: str

93
libervia/backend/plugins/plugin_misc_identity.py
View file

@ -22,12 +22,15 @@ from base64 import b64encode
import hashlib
from typing import Any, Coroutine, Dict, List, Optional, Union, cast
from pydantic import BaseModel, Field, RootModel
from twisted.internet import defer, reactor, threads
from twisted.internet.interfaces import IReactorCore
from twisted.words.protocols.jabber import jid
from twisted.words.protocols.jabber.error import StanzaError
from libervia.backend.core import exceptions
from libervia.backend.models.types import JIDType
from libervia.backend.core.constants import Const as C
from libervia.backend.core.i18n import _
from libervia.backend.core.log import getLogger
@ -59,7 +62,7 @@ PLUGIN_INFO = {
C.PI_PROTOCOLS: [],
C.PI_DEPENDENCIES: [],
C.PI_RECOMMENDATIONS: ["XEP-0045"],
C.PI_MAIN: "Identity",
C.PI_MAIN: "IdentityPlugin",
C.PI_HANDLER: "no",
C.PI_DESCRIPTION: _("""Identity manager"""),
}
@ -68,7 +71,45 @@ Callback = namedtuple("Callback", ("origin", "get", "set", "priority"))
AVATAR_DIM = (128, 128)
class Identity:
class EntityType(BaseModel):
category: str = Field(
description='Entity category (e.g. "client", "conference", "gateway").'
)
type: str = Field(description='Entity type (e.g. "pc", "text", "irc").')
name: str | None = Field(None, description="Human-readable name of the entity.")
lang: str | None = Field(None, description="Language tag of the name.")
class IdentityData(BaseModel):
types: list[EntityType] = Field(
default_factory=list, description="Disco identity types."
)
avatar: str | None = Field(None, description="Avatar hash.")
nicknames: list[str] = Field(
default_factory=list, description="Known nicknames of the entity."
)
class Identities(RootModel):
root: dict[JIDType, IdentityData]
def items(self):
return self.root.items()
def __dict__(self) -> dict[JIDType, IdentityData]: # type: ignore
return self.root
def __iter__(self): # type: ignore
return iter(self.root)
def __getitem__(self, item):
return self.root[item]
def __setitem__(self, key: jid.JID, value: IdentityData) -> None:
self.root[key] = value
class IdentityPlugin:
def __init__(self, host):
log.info(_("Plugin Identity initialization"))
@ -715,7 +756,7 @@ class Identity:
d = defer.ensureDeferred(
self.get_identity(client, entity, metadata_filter, use_cache)
)
d.addCallback(data_format.serialise)
d.addCallback(lambda identity: identity.model_dump_json())
return d
async def get_identity(
@ -724,7 +765,7 @@ class Identity:
entity: Optional[jid.JID] = None,
metadata_filter: Optional[List[str]] = None,
use_cache: bool = True,
) -> Dict[str, Any]:
) -> IdentityData:
"""Retrieve identity of an entity
@param entity: entity to check
@ -732,10 +773,9 @@ class Identity:
metadata in this filter
@param use_cache: if False, cache won't be checked
should be True most of time, to avoid useless network requests
@return: identity data
@return: IdentityData with disco identities and metadata
"""
identities = []
id_data = {"identities": identities}
identity = IdentityData()
if not metadata_filter:
metadata_names = self.metadata.keys()
@ -743,9 +783,11 @@ class Identity:
metadata_names = metadata_filter
for metadata_name in metadata_names:
id_data[metadata_name] = await self.get(
client, metadata_name, entity, use_cache
)
value = await self.get(client, metadata_name, entity, use_cache)
if metadata_name == "avatar":
identity.avatar = value
elif metadata_name == "nicknames":
identity.nicknames = value or []
if entity is not None:
try:
@ -760,23 +802,22 @@ class Identity:
log.exception(f"Can't get disco infos for {entity}.")
else:
for (category, type_), identities_map in disco_infos.identities.items():
for lang, identity in identities_map.items():
identity_entry = {
"category": identity.category,
"type": identity.type,
"name": identity.name,
}
if lang is not None:
identity_entry["lang"] = lang
identities.append(identity_entry)
return id_data
for lang, disco_identity in identities_map.items():
identity.types.append(
EntityType(
category=disco_identity.category,
type=disco_identity.type,
name=disco_identity.name,
lang=lang,
)
)
return identity
def _get_identities(self, entities_s, metadata_filter, profile):
entities = [jid.JID(e) for e in entities_s]
client = self.host.get_client(profile)
d = defer.ensureDeferred(self.get_identities(client, entities, metadata_filter))
d.addCallback(lambda d: data_format.serialise({str(j): i for j, i in d.items()}))
d.addCallback(lambda identities: identities.model_dump_json(exclude_none=True))
return d
async def get_identities(
@ -784,7 +825,7 @@ class Identity:
client: SatXMPPEntity,
entities: List[jid.JID],
metadata_filter: Optional[List[str]] = None,
) -> dict:
) -> Identities:
"""Retrieve several identities at once
@param entities: entities from which identities must be retrieved
@ -793,7 +834,7 @@ class Identity:
if an error happens while retrieve a jid entity, it won't be present in the
result (and a warning will be logged)
"""
identities = {}
identities = Identities({})
get_identity_list = []
for entity_jid in entities:
get_identity_list.append(
@ -819,13 +860,13 @@ class Identity:
def _get_base_identities(self, profile_key):
client = self.host.get_client(profile_key)
d = defer.ensureDeferred(self.get_base_identities(client))
d.addCallback(lambda d: data_format.serialise({str(j): i for j, i in d.items()}))
d.addCallback(lambda identities: identities.model_dump_json(exclude_none=True))
return d
async def get_base_identities(
self,
client: SatXMPPEntity,
) -> dict:
) -> Identities:
"""Retrieve identities for entities in roster + own identity + invitations
@param with_guests: if True, get affiliations of people invited by email