Retro Bulletin Board Systems on atproto. Web app and TUI.
lazy mirror of alyraffauf/atbbs
atbbs.xyz
forums
python
tui
atproto
bbs
tui: abtract compose screens
+138
-184
+1
-3
tui/app.tcss
+1
-3
tui/app.tcss
+98
tui/screens/compose/base.py
+98
tui/screens/compose/base.py
···
1
+
"""Base class for all compose screens (thread, news, reply).
2
+
3
+
Handles the shared post workflow: session check, body/title validation,
4
+
file upload, API call, error handling, and navigation on success.
5
+
6
+
Subclasses define their own ``compose()`` layout using these widget IDs:
7
+
- ``#compose-title`` — optional Input (only if ``requires_title = True``)
8
+
- ``#compose-body`` — required TextArea
9
+
- ``#compose-file`` — optional Input for a file path
10
+
11
+
Subclasses must override ``get_post_params()`` to return the keyword
12
+
arguments passed to ``create_post_record``.
13
+
"""
14
+
15
+
from textual import work
16
+
from textual.screen import Screen
17
+
from textual.widgets import Input, TextArea
18
+
19
+
from core import limits
20
+
from core.models import AuthError
21
+
from core.records import create_post_record
22
+
from tui.screens.compose.upload import upload_file
23
+
from tui.util import require_session
24
+
25
+
26
+
class ComposeScreen(Screen):
27
+
BINDINGS = [
28
+
("escape", "app.pop_screen", "back"),
29
+
("ctrl+s", "post", "post"),
30
+
]
31
+
32
+
# Subclasses set these to control validation behavior.
33
+
requires_title: bool = False
34
+
post_type: str = "post" # used in error messages, e.g. "Failed to post thread"
35
+
36
+
def get_post_params(self, title: str | None, body: str) -> dict:
37
+
"""Return keyword arguments for ``create_post_record``.
38
+
39
+
Called after validation passes. *title* is ``None`` when
40
+
``requires_title`` is ``False``.
41
+
"""
42
+
raise NotImplementedError
43
+
44
+
def action_post(self) -> None:
45
+
self._do_post()
46
+
47
+
@work(exclusive=True)
48
+
async def _do_post(self) -> None:
49
+
session = require_session(self)
50
+
if not session:
51
+
return
52
+
53
+
# -- Read form values ------------------------------------------------
54
+
title = None
55
+
if self.requires_title:
56
+
title = self.query_one("#compose-title", Input).value.strip()
57
+
58
+
body = self.query_one("#compose-body", TextArea).text.strip()
59
+
60
+
# -- Validate --------------------------------------------------------
61
+
if self.requires_title and not title:
62
+
self.notify("Title cannot be empty.", severity="error")
63
+
return
64
+
if not body:
65
+
self.notify("Body cannot be empty.", severity="error")
66
+
return
67
+
if len(body) > limits.POST_BODY:
68
+
self.notify(
69
+
f"Body too long ({len(body)}/{limits.POST_BODY}).", severity="error"
70
+
)
71
+
return
72
+
73
+
# -- Upload attachment (if any) --------------------------------------
74
+
attachments = []
75
+
file_path = self.query_one("#compose-file", Input).value.strip()
76
+
if file_path:
77
+
attachments = await upload_file(self, file_path, session)
78
+
if attachments is None:
79
+
return # upload_file already notified the user
80
+
81
+
# -- Create the post record ------------------------------------------
82
+
params = self.get_post_params(title, body)
83
+
try:
84
+
resp = await create_post_record(
85
+
self.app.http_client,
86
+
session,
87
+
attachments=attachments or None,
88
+
**params,
89
+
)
90
+
resp.raise_for_status()
91
+
except AuthError:
92
+
self.notify("Session expired. Please log in again.", severity="error")
93
+
return
94
+
except Exception as error:
95
+
self.notify(f"Failed to post {self.post_type}: {error}", severity="error")
96
+
return
97
+
98
+
self.app.pop_screen()
+11
-61
tui/screens/compose/news.py
+11
-61
tui/screens/compose/news.py
···
1
-
from textual import work
2
1
from textual.app import ComposeResult
3
2
from textual.containers import Vertical
4
-
from textual.screen import Screen
5
3
from textual.widgets import Footer, Input, Static, TextArea
6
4
7
5
from core import lexicon, limits
8
-
from core.models import AtUri, AuthError, BBS
9
-
from core.records import create_post_record
10
-
from tui.util import require_session
6
+
from core.models import AtUri, BBS
7
+
from tui.screens.compose.base import ComposeScreen
11
8
from tui.widgets.breadcrumb import Breadcrumb
12
-
from tui.screens.compose.upload import upload_file
13
9
14
10
15
-
class ComposeNewsScreen(Screen):
16
-
BINDINGS = [
17
-
("escape", "app.pop_screen", "back"),
18
-
("ctrl+s", "post", "post"),
19
-
]
11
+
class ComposeNewsScreen(ComposeScreen):
12
+
requires_title = True
13
+
post_type = "news"
20
14
21
15
def __init__(self, bbs: BBS, handle: str) -> None:
22
16
super().__init__()
···
32
26
with Vertical():
33
27
yield Static("news", classes="title")
34
28
yield Input(
35
-
placeholder="Title", id="news-title", max_length=limits.POST_TITLE
29
+
placeholder="Title", id="compose-title", max_length=limits.POST_TITLE
36
30
)
37
-
yield TextArea(id="news-body", language=None)
38
-
yield Input(placeholder="attach file (path, optional)", id="news-file")
31
+
yield TextArea(id="compose-body", language=None)
32
+
yield Input(placeholder="attach file (path, optional)", id="compose-file")
39
33
yield Footer()
40
34
41
35
def on_mount(self) -> None:
42
-
self.query_one("#news-title", Input).focus()
43
-
44
-
def action_post(self) -> None:
45
-
self.post_news()
46
-
47
-
@work(exclusive=True)
48
-
async def post_news(self) -> None:
49
-
session = require_session(self)
50
-
if not session:
51
-
return
52
-
53
-
title = self.query_one("#news-title", Input).value.strip()
54
-
body = self.query_one("#news-body", TextArea).text.strip()
55
-
if not title or not body:
56
-
self.notify("Title and body cannot be empty.", severity="error")
57
-
return
58
-
if len(body) > limits.POST_BODY:
59
-
self.notify(
60
-
f"Body too long ({len(body)}/{limits.POST_BODY}).", severity="error"
61
-
)
62
-
return
36
+
self.query_one("#compose-title", Input).focus()
63
37
38
+
def get_post_params(self, title: str | None, body: str) -> dict:
64
39
site_uri = str(AtUri(self.bbs.identity.did, lexicon.SITE, "self"))
65
-
66
-
attachments = []
67
-
file_path = self.query_one("#news-file", Input).value.strip()
68
-
if file_path:
69
-
attachments = await upload_file(self, file_path, session)
70
-
if attachments is None:
71
-
return
72
-
73
-
try:
74
-
resp = await create_post_record(
75
-
self.app.http_client,
76
-
session,
77
-
scope=site_uri,
78
-
body=body,
79
-
title=title,
80
-
attachments=attachments or None,
81
-
)
82
-
resp.raise_for_status()
83
-
except AuthError:
84
-
self.notify("Session expired. Please log in again.", severity="error")
85
-
return
86
-
except Exception as error:
87
-
self.notify(f"Failed to post news: {error}", severity="error")
88
-
return
89
-
90
-
self.app.pop_screen()
40
+
return {"scope": site_uri, "body": body, "title": title}
+17
-59
tui/screens/compose/reply.py
+17
-59
tui/screens/compose/reply.py
···
1
-
from textual import work
2
1
from textual.app import ComposeResult
3
2
from textual.containers import Vertical
4
-
from textual.screen import Screen
5
3
from textual.widgets import Footer, Input, Static, TextArea
6
4
7
-
from core import limits
8
-
from core.models import AuthError, BBS, Post as PostModel
9
-
from core.records import create_post_record
10
-
from tui.util import require_session
5
+
from core.models import BBS, Post as PostModel
6
+
from tui.screens.compose.base import ComposeScreen
11
7
from tui.widgets.breadcrumb import Breadcrumb
12
-
from tui.screens.compose.upload import upload_file
13
8
14
9
15
-
class ComposeReplyScreen(Screen):
10
+
class ComposeReplyScreen(ComposeScreen):
11
+
requires_title = False
12
+
post_type = "reply"
13
+
16
14
BINDINGS = [
17
-
("escape", "app.pop_screen", "back"),
18
-
("ctrl+s", "post", "post"),
19
15
("ctrl+g", "toggle_reply_to", "toggle reply to"),
20
16
]
21
17
···
47
43
classes="subtitle",
48
44
id="reply-to-info",
49
45
)
50
-
yield TextArea(id="reply-body", language=None)
51
-
yield Input(placeholder="attach file (path, optional)", id="reply-file")
46
+
yield TextArea(id="compose-body", language=None)
47
+
yield Input(placeholder="attach file (path, optional)", id="compose-file")
52
48
yield Footer()
53
49
54
50
def on_mount(self) -> None:
55
-
self.query_one("#reply-body", TextArea).focus()
51
+
self.query_one("#compose-body", TextArea).focus()
56
52
57
53
def action_toggle_reply_to(self) -> None:
58
54
if not self.original_parent:
···
73
69
classes="subtitle",
74
70
id="reply-to-info",
75
71
),
76
-
before=self.query_one("#reply-body"),
77
-
)
78
-
79
-
def action_post(self) -> None:
80
-
self.post_reply()
81
-
82
-
@work(exclusive=True)
83
-
async def post_reply(self) -> None:
84
-
session = require_session(self)
85
-
if not session:
86
-
return
87
-
88
-
body = self.query_one("#reply-body", TextArea).text.strip()
89
-
if not body:
90
-
self.notify("Message body cannot be empty.", severity="error")
91
-
return
92
-
if len(body) > limits.POST_BODY:
93
-
self.notify(
94
-
f"Body too long ({len(body)}/{limits.POST_BODY}).", severity="error"
95
-
)
96
-
return
97
-
98
-
attachments = []
99
-
file_path = self.query_one("#reply-file", Input).value.strip()
100
-
if file_path:
101
-
attachments = await upload_file(self, file_path, session)
102
-
if attachments is None:
103
-
return
104
-
105
-
try:
106
-
resp = await create_post_record(
107
-
self.app.http_client,
108
-
session,
109
-
scope=self.thread.scope,
110
-
body=body,
111
-
root=self.thread.uri,
112
-
parent=self.parent_post.uri if self.parent_post else None,
113
-
attachments=attachments or None,
72
+
before=self.query_one("#compose-body"),
114
73
)
115
-
resp.raise_for_status()
116
-
except AuthError:
117
-
self.notify("Session expired. Please log in again.", severity="error")
118
-
return
119
-
except Exception as error:
120
-
self.notify(f"Failed to post reply: {error}", severity="error")
121
-
return
122
74
123
-
self.app.pop_screen()
75
+
def get_post_params(self, title: str | None, body: str) -> dict:
76
+
return {
77
+
"scope": self.thread.scope,
78
+
"body": body,
79
+
"root": self.thread.uri,
80
+
"parent": self.parent_post.uri if self.parent_post else None,
81
+
}
+11
-61
tui/screens/compose/thread.py
+11
-61
tui/screens/compose/thread.py
···
1
-
from textual import work
2
1
from textual.app import ComposeResult
3
2
from textual.containers import Vertical
4
-
from textual.screen import Screen
5
3
from textual.widgets import Footer, Input, Static, TextArea
6
4
7
5
from core import lexicon, limits
8
-
from core.models import AtUri, AuthError, BBS, Board
9
-
from core.records import create_post_record
10
-
from tui.util import require_session
6
+
from core.models import AtUri, BBS, Board
7
+
from tui.screens.compose.base import ComposeScreen
11
8
from tui.widgets.breadcrumb import Breadcrumb
12
-
from tui.screens.compose.upload import upload_file
13
9
14
10
15
-
class ComposeThreadScreen(Screen):
16
-
BINDINGS = [
17
-
("escape", "app.pop_screen", "back"),
18
-
("ctrl+s", "post", "post"),
19
-
]
11
+
class ComposeThreadScreen(ComposeScreen):
12
+
requires_title = True
13
+
post_type = "thread"
20
14
21
15
def __init__(self, bbs: BBS, handle: str, board: Board) -> None:
22
16
super().__init__()
···
35
29
yield Static("new thread", classes="title")
36
30
yield Input(
37
31
placeholder="Thread title",
38
-
id="thread-title",
32
+
id="compose-title",
39
33
max_length=limits.POST_TITLE,
40
34
)
41
-
yield TextArea(id="thread-body", language=None)
42
-
yield Input(placeholder="attach file (path, optional)", id="thread-file")
35
+
yield TextArea(id="compose-body", language=None)
36
+
yield Input(placeholder="attach file (path, optional)", id="compose-file")
43
37
yield Footer()
44
38
45
39
def on_mount(self) -> None:
46
-
self.query_one("#thread-title", Input).focus()
47
-
48
-
def action_post(self) -> None:
49
-
self.post_thread()
50
-
51
-
@work(exclusive=True)
52
-
async def post_thread(self) -> None:
53
-
session = require_session(self)
54
-
if not session:
55
-
return
56
-
57
-
title = self.query_one("#thread-title", Input).value.strip()
58
-
body = self.query_one("#thread-body", TextArea).text.strip()
59
-
if not title or not body:
60
-
self.notify("Title and body cannot be empty.", severity="error")
61
-
return
62
-
if len(body) > limits.POST_BODY:
63
-
self.notify(
64
-
f"Body too long ({len(body)}/{limits.POST_BODY}).", severity="error"
65
-
)
66
-
return
40
+
self.query_one("#compose-title", Input).focus()
67
41
42
+
def get_post_params(self, title: str | None, body: str) -> dict:
68
43
board_uri = str(AtUri(self.bbs.identity.did, lexicon.BOARD, self.board.slug))
69
-
70
-
attachments = []
71
-
file_path = self.query_one("#thread-file", Input).value.strip()
72
-
if file_path:
73
-
attachments = await upload_file(self, file_path, session)
74
-
if attachments is None:
75
-
return
76
-
77
-
try:
78
-
resp = await create_post_record(
79
-
self.app.http_client,
80
-
session,
81
-
scope=board_uri,
82
-
body=body,
83
-
title=title,
84
-
attachments=attachments or None,
85
-
)
86
-
resp.raise_for_status()
87
-
except AuthError:
88
-
self.notify("Session expired. Please log in again.", severity="error")
89
-
return
90
-
except Exception as error:
91
-
self.notify(f"Failed to post thread: {error}", severity="error")
92
-
return
93
-
94
-
self.app.pop_screen()
44
+
return {"scope": board_uri, "body": body, "title": title}