zenfyr.dev
social media crossposting tool. 3rd time's the charm
mastodon
misskey
crossposting
bluesky
add bsky output skeleton
+53
+50
bluesky/output.py
+50
bluesky/output.py
···
1
+
from dataclasses import dataclass
2
+
from typing import Any, override
3
+
4
+
from cross.post import Post, PostRef
5
+
from cross.service import OutputService
6
+
from database.connection import DatabasePool
7
+
from bluesky.info import SERVICE, BlueskyService, validate_and_transform
8
+
9
+
10
+
@dataclass(kw_only=True)
11
+
class BlueskyOutputOptions:
12
+
handle: str | None = None
13
+
did: str | None = None
14
+
pds: str | None = None
15
+
16
+
@classmethod
17
+
def from_dict(cls, data: dict[str, Any]) -> "BlueskyOutputOptions":
18
+
validate_and_transform(data)
19
+
return BlueskyOutputOptions(**data)
20
+
21
+
22
+
class BlueskyOutputService(BlueskyService, OutputService):
23
+
def __init__(self, db: DatabasePool, options: BlueskyOutputOptions) -> None:
24
+
super().__init__(SERVICE, db)
25
+
self.options: BlueskyOutputOptions = options
26
+
self._init_identity()
27
+
28
+
@override
29
+
def get_identity_options(self) -> tuple[str | None, str | None, str | None]:
30
+
return (self.options.handle, self.options.did, self.options.pds)
31
+
32
+
@override
33
+
def accept_post(self, post: Post):
34
+
# TODO: post creation
35
+
self.log.warning("NOT IMPLEMENTED: accept_post %s", post.id)
36
+
37
+
@override
38
+
def delete_post(self, post: PostRef):
39
+
# TODO: post deletion
40
+
self.log.warning("NOT IMPLEMENTED: delete_post %s", post.id)
41
+
42
+
@override
43
+
def accept_repost(self, repost: PostRef, reposted: PostRef):
44
+
# TODO: reposts
45
+
self.log.warning("NOT IMPLEMENTED: accept_repost %s of %s", repost.id, reposted.id)
46
+
47
+
@override
48
+
def delete_repost(self, repost: PostRef):
49
+
# TODO: unreposts
50
+
self.log.warning("NOT IMPLEMENTED: delete_repost %s", repost.id)