start centralizing id conversion into new ids.py file

ids.py

@@ -0,0 +1,46 @@
+"""Convert ids and handles between protocols.
+
+https://fed.brid.gy/docs#translate
+"""
+from granary.bluesky import Bluesky
+
+from activitypub import ActivityPub
+from atproto import ATProto
+from common import host_url
+from models import User
+from protocol import Protocol, PROTOCOLS
+from web import Web
+
+
+def convert_id(*, id, from_proto, to_proto):
+    """Converts an id (not necessarily handle) from one protocol to another.
+
+    Args:
+      id (str)
+      from_proto (:class:`Protocol`)
+      to_proto (:class:`Protocol`)
+
+    Returns:
+      str: the corresponding id in ``to_proto``
+    """
+    assert id and from_proto and to_proto
+    assert from_proto != to_proto
+
+    match (from_proto.LABEL, to_proto.LABEL):
+        case (_, 'atproto'):
+            user = from_proto.get_by_id(id)
+            return user.atproto_did if user else None
+        case ('atproto', _):
+            user = from_proto.get_for_copy(id)
+            return user.key.id() if user else None
+        case (_, 'activitypub'):
+            return host_url(f'{from_proto.ABBREV}/{ActivityPub.ABBREV}/{id}')
+        case ('activitypub', 'web'):
+            return id
+        # fake protocol is only for unit tests
+        case (_, 'fake'):
+            return f'fake:{id}'
+        case ('fake', _):
+            return id
+
+    assert False

models.py

@@ -181,6 +181,17 @@ class User(StringIdModel, metaclass=ProtocolUserMeta):
 
         return user
 
+    @staticmethod
+    def get_for_copy(copy_id):
+        """Fetches a user with a given id in copies.
+
+        Thin wrapper around :meth:User.get_copies` that returns the first
+        matching :class:`User`.
+        """
+        users = User.get_for_copies([copy_id])
+        if users:
+            return users[0]
+
     @staticmethod
     def get_for_copies(copy_ids):
         """Fetches users (across all protocols) for a given set of copies.
@@ -518,9 +529,8 @@ class Object(StringIdModel):
             logger.info(f'Replacing {owner_field} {obj.get(owner_field)}...')
 
             # load matching user, if any
-            users = User.get_for_copies([owner])
-            if users:
-                user = users[0]
+            user = User.get_for_copy(owner)
+            if user:
                 if user.obj and user.obj.as1:
                     obj[owner_field] = {
                         **user.obj.as1,

tests/test_ids.py

@@ -0,0 +1,44 @@
+"""Unit tests for ids.py."""
+from activitypub import ActivityPub
+from atproto import ATProto
+from ids import convert_id
+from models import Target
+from web import Web
+from .testutil import Fake, TestCase
+
+
+class IdsTest(TestCase):
+    def test_convert_id(self):
+        Web(id='user.com', atproto_did='did:plc:123',
+            copies=[Target(uri='did:plc:123', protocol='atproto')]).put()
+        ActivityPub(id='https://server/user', atproto_did='did:plc:456',
+                    copies=[Target(uri='did:plc:456', protocol='atproto')]).put()
+        Fake(id='fake:user', atproto_did='did:plc:789',
+             copies=[Target(uri='did:plc:789', protocol='atproto')]).put()
+
+        for from_, id, to, expected in [
+            (Web, 'user.com', ActivityPub, 'http://localhost/web/ap/user.com'),
+            (Web, 'user.com', ATProto, 'did:plc:123'),
+            (Web, 'user.com', Fake, 'fake:user.com'),
+                # TODO: not a domain, is that ok?
+            (ActivityPub, 'https://server/user', Web, 'https://server/user'),
+            (ActivityPub, 'https://server/user', ATProto, 'did:plc:456'),
+            (ActivityPub, 'https://server/user', Fake, 'fake:https://server/user'),
+            (ATProto, 'did:plc:123', Web, 'user.com'),
+            (ATProto, 'did:plc:456', ActivityPub, 'https://server/user'),
+            (ATProto, 'did:plc:789', Fake, 'fake:user'),
+            (Fake, 'fake:user', Web, 'fake:user'),
+            (Fake, 'fake:user', ActivityPub, 'http://localhost/fa/ap/fake:user'),
+            (Fake, 'fake:user', ATProto, 'did:plc:789'),
+        ]:
+            with self.subTest(from_=from_.LABEL, to=to.LABEL):
+                self.assertEqual(expected, convert_id(
+                    id=id, from_proto=from_, to_proto=to))
+
+    def test_convert_id_no_atproto_did_stored(self):
+        for proto in Web, ActivityPub, Fake:
+            with self.subTest(proto=proto.LABEL):
+                self.assertIsNone(convert_id(
+                    id='foo', from_proto=proto, to_proto=ATProto))
+                self.assertIsNone(convert_id(
+                    id='did:plc:123', from_proto=ATProto, to_proto=proto))