jmbwell/rsbbs
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

move data logic to plugins, split models from controller

Browse Source
This commit is contained in:
John Burwell 2023-04-27 16:51:09 -05:00
parent 52aeb1128f
commit 60293152d6
16 changed files with 220 additions and 207 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
rsbbs/config.py
View File

@ -48,6 +48,7 @@ class Config():
def __getattr__(self, __name: str):
return self.config[__name]
# Format the config for display
def __repr__(self):
repr = []
repr.append(f"app_name: {self.app_name}\r\n")

19
rsbbs/console.py
View File

@ -16,11 +16,8 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import os
import sys
import sqlalchemy.exc
import rsbbs
from rsbbs.config import Config
from rsbbs.controller import Controller
@ -134,22 +131,6 @@ class Console():
f"{datetime_: <{11}} "
f"{message.Message.subject}")
#
# Command functions
#
def send_private(self, args):
self.send(args, is_private=True)
"""Send a message visible only to the recipient callsign.
Required arguments:
callsign -- the recipient's callsign
Optional arguments:
subject -- message subject
message -- the message itself
"""
#
# Main input loop
#

134
rsbbs/controller.py
View File

@ -16,42 +16,20 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import subprocess
from datetime import datetime, timezone
from sqlalchemy import Boolean, DateTime, String
from sqlalchemy import create_engine, delete, select, or_
from sqlalchemy.orm import DeclarativeBase, Mapped, Session
from sqlalchemy.orm import mapped_column
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from rsbbs.config import Config
class Base(DeclarativeBase):
pass
class Message(Base):
__tablename__ = 'message'
id: Mapped[int] = mapped_column(primary_key=True)
sender: Mapped[str] = mapped_column(String)
recipient: Mapped[str] = mapped_column(String)
subject: Mapped[str] = mapped_column(String)
message: Mapped[str] = mapped_column(String)
datetime: Mapped[DateTime] = mapped_column(
DateTime, default=datetime.now(timezone.utc))
is_private: Mapped[bool] = mapped_column(Boolean)
from rsbbs.models import Base
class Controller():
def __init__(self, config: Config):
def __init__(self, config: Config) -> None:
self.config = config
self._init_datastore()
def _init_datastore(self):
def _init_datastore(self) -> None:
"""Create a connection to the sqlite3 database.
The default location is the system-specific user-level data directory.
@ -64,103 +42,5 @@ class Controller():
# Create the database schema if none exists
Base.metadata.create_all(self.engine)
def delete(self, args):
"""Delete a message.
Arguments:
number -- the message number to delete
"""
with Session(self.engine) as session:
try:
message = session.get(Message, args.number)
session.delete(message)
session.commit()
except Exception:
raise
def delete_mine(self, args):
"""Delete all messages addressed to the calling station's callsign."""
with Session(self.engine) as session:
try:
statement = delete(Message).where(
Message.recipient == self.config.calling_station
).returning(Message)
result = session.execute(
statement,
execution_options={"prebuffer_rows": True})
session.commit()
return result
except Exception:
raise
def list(self, args):
"""List all messages."""
with Session(self.engine) as session:
try:
# Using or_ and is_ etc. to distinguish from python operators
statement = select(Message).where(
or_(
(Message.is_private.is_(False)),
(Message.recipient.__eq__(
self.config.calling_station)))
)
result = session.execute(
statement,
execution_options={"prebuffer_rows": True})
except Exception:
raise
return result
def list_mine(self, args):
"""List only messages addressed to the calling station's callsign,
including public and private messages.
"""
with Session(self.engine) as session:
try:
statement = select(Message).where(
Message.recipient == self.config.calling_station)
result = session.execute(
statement,
execution_options={"prebuffer_rows": True})
return result
except Exception:
raise
def read(self, args):
"""Read a message.
Arguments:
number -- the message number to read
"""
with Session(self.engine) as session:
try:
statement = select(Message).where(Message.id == args.number)
result = session.execute(statement).one()
return result
except Exception:
raise
def send(self, args, is_private=False):
"""Create a new message addressed to another callsign.
Required arguments:
callsign -- the recipient's callsign
Optional arguments:
subject -- message subject
message -- the message itself
"""
with Session(self.engine) as session:
try:
session.add(Message(
sender=self.config.calling_station.upper(),
recipient=args.callsign.upper(),
subject=args.subject,
message=args.message,
is_private=is_private
))
session.commit()
return {}
except Exception:
session.rollback()
raise
def session(self) -> Session:
return Session(self.engine)

40
rsbbs/models.py Normal file
View File

@ -0,0 +1,40 @@
#!/usr/bin/env python
#
# Really Simple BBS - a really simple BBS for ax.25 packet radio.
# Copyright (C) 2023 John Burwell <john@atatdotdot.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
from datetime import datetime, timezone
from sqlalchemy import Boolean, DateTime, String
from sqlalchemy.orm import DeclarativeBase, Mapped
from sqlalchemy.orm import mapped_column
class Base(DeclarativeBase):
pass
class Message(Base):
__tablename__ = 'message'
id: Mapped[int] = mapped_column(primary_key=True)
sender: Mapped[str] = mapped_column(String)
recipient: Mapped[str] = mapped_column(String)
subject: Mapped[str] = mapped_column(String)
message: Mapped[str] = mapped_column(String)
datetime: Mapped[DateTime] = mapped_column(
DateTime, default=datetime.now(timezone.utc))
is_private: Mapped[bool] = mapped_column(Boolean)

6
rsbbs/plugins/bye/plugin.py
View File

@ -22,20 +22,20 @@ from rsbbs.parser import Parser
class Plugin():
def __init__(self, api: Console):
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
if api.config.debug:
print(f"Plugin {__name__} loaded")
def init_parser(self, parser: Parser):
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='bye',
aliases=['b', 'q'],
help='Sign off and disconnect')
subparser.set_defaults(func=self.run)
def run(self, args):
def run(self, args) -> None:
"""Disconnect and exit."""
self.api.write_output("Bye!")
exit(0)

32
rsbbs/plugins/delete/plugin.py
View File

@ -16,19 +16,23 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import sqlalchemy
import sqlalchemy.exc
from rsbbs.console import Console
from rsbbs.models import Message
from rsbbs.parser import Parser
class Plugin():
def __init__(self, api: Console):
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
if api.config.debug:
print(f"Plugin {__name__} loaded")
def init_parser(self, parser: Parser):
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='delete',
aliases=['d', 'k'],
@ -37,11 +41,23 @@ class Plugin():
help='The number of the message to delete')
subparser.set_defaults(func=self.run)
def run(self, args):
"""Delete a message specified by ID number."""
if args.number:
def delete(self, number) -> None:
with self.api.controller.session() as session:
try:
self.api.controller.delete(args)
self.api.write_output(f"Deleted message #{args.number}")
except Exception as e:
message = session.get(Message, number)
session.delete(message)
session.commit()
self.api.write_output(f"Deleted message #{number}")
except sqlalchemy.exc.NoResultFound:
self.api.write_output(f"Message not found.")
except Exception as e:
print(e)
def run(self, args) -> None:
"""Delete a message.
Arguments:
number -- the message number to delete
"""
if args.number:
self.delete(args.number)

31
rsbbs/plugins/deletem/plugin.py
View File

@ -16,34 +16,38 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import sqlalchemy
from rsbbs.console import Console
from rsbbs.models import Message
from rsbbs.parser import Parser
class Plugin():
def __init__(self, api: Console):
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
if api.config.debug:
print(f"Plugin {__name__} loaded")
def init_parser(self, parser: Parser):
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='deletem',
aliases=['dm', 'km'],
help='Delete all messages addressed to you')
subparser.set_defaults(func=self.run)
def run(self, args):
"""Delete all messages addressed to the calling station's callsign."""
response = self.api.read_line(
"Delete all messages addressed to you? Y/N:")
if response.lower() != "y":
return
else:
def delete_mine(self) -> None:
with self.api.controller.session() as session:
try:
result = self.api.controller.delete_mine(args)
statement = sqlalchemy.delete(Message).where(
Message.recipient == self.api.config.calling_station
).returning(Message)
result = session.execute(
statement,
execution_options={"prebuffer_rows": True})
session.commit()
messages = result.all()
count = len(messages)
if count > 0:
@ -52,3 +56,10 @@ class Plugin():
self.api.write_output(f"No messages to delete.")
except Exception as e:
self.api.write_output(f"Unable to delete messages: {e}")
def run(self, args) -> None:
"""Delete all messages addressed to the calling station's callsign."""
response = self.api.read_line(
"Delete all messages addressed to you? Y/N:")
if response.lower() == "y":
self.delete_mine()

6
rsbbs/plugins/heard/plugin.py
View File

@ -24,20 +24,20 @@ from rsbbs.parser import Parser
class Plugin():
def __init__(self, api: Console):
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
if api.config.debug:
print(f"Plugin {__name__} loaded")
def init_parser(self, parser: Parser):
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='heard',
aliases=['j'],
help='Show heard stations log')
subparser.set_defaults(func=self.run)
def run(self, args):
def run(self, args) -> None:
"""Show a log of stations that have been heard by this station,
also known as the 'mheard' (linux) or 'jheard' (KPC, etc.) log.
"""

6
rsbbs/plugins/help/plugin.py
View File

@ -22,20 +22,20 @@ from rsbbs.parser import Parser
class Plugin():
def __init__(self, api: Console):
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
if api.config.debug:
print(f"Plugin {__name__} loaded")
def init_parser(self, parser: Parser):
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='help',
aliases=['h', '?'],
help='Show help')
subparser.set_defaults(func=self.run)
def run(self, args):
def run(self, args) -> None:
"""Show a log of stations that have been heard by this station,
also known as the 'mheard' (linux) or 'jheard' (KPC, etc.) log.
"""

29
rsbbs/plugins/list/plugin.py
View File

@ -16,26 +16,47 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import sqlalchemy
from rsbbs.console import Console
from rsbbs.models import Message
from rsbbs.parser import Parser
class Plugin():
def __init__(self, api: Console):
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
if api.config.debug:
print(f"Plugin {__name__} loaded")
def init_parser(self, parser: Parser):
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='list',
aliases=['l'],
help='List all available messages')
subparser.set_defaults(func=self.run)
def run(self, args):
def list(self, args) -> sqlalchemy.ChunkedIteratorResult:
"""List all messages."""
with self.api.controller.session() as session:
try:
# Using or_ and is_ etc. to distinguish from python operators
statement = sqlalchemy.select(Message).where(
sqlalchemy.or_(
(Message.is_private.is_(False)),
(Message.recipient.__eq__(
self.api.config.calling_station)))
)
result = session.execute(
statement,
execution_options={"prebuffer_rows": True})
except Exception:
raise
return result
def run(self, args) -> None:
"""List all public messages and messages private to the caller."""
result = self.api.controller.list(args)
result = self.list(args)
self.api.print_message_list(result)

21
rsbbs/plugins/listm/plugin.py
View File

@ -16,28 +16,43 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import sqlalchemy
from rsbbs.console import Console
from rsbbs.models import Message
from rsbbs.parser import Parser
class Plugin():
def __init__(self, api: Console):
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
if api.config.debug:
print(f"Plugin {__name__} loaded")
def init_parser(self, parser: Parser):
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='listm',
aliases=['lm'],
help='List only messages addressed to you')
subparser.set_defaults(func=self.run)
def list_mine(self, args) -> sqlalchemy.ChunkedIteratorResult:
with self.api.controller.session() as session:
try:
statement = sqlalchemy.select(Message).where(
Message.recipient == self.api.config.calling_station)
result = session.execute(
statement,
execution_options={"prebuffer_rows": True})
return result
except Exception:
raise
def run(self, args):
"""List only messages addressed to the calling station's callsign,
including public and private messages.
"""
result = self.api.controller.list_mine(args)
result = self.list_mine(args)
self.api.print_message_list(result)

28
rsbbs/plugins/read/plugin.py
View File

@ -17,20 +17,22 @@
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import sqlalchemy
import sqlalchemy.exc
from rsbbs.console import Console
from rsbbs.models import Message
from rsbbs.parser import Parser
class Plugin():
def __init__(self, api: Console):
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
if api.config.debug:
print(f"Plugin {__name__} loaded")
def init_parser(self, parser: Parser):
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='read',
aliases=['r'],
@ -38,17 +40,23 @@ class Plugin():
subparser.add_argument('number', help='Message number to read')
subparser.set_defaults(func=self.run)
def run(self, args):
def read_message(self, number) -> None:
with self.api.controller.session() as session:
try:
statement = sqlalchemy.select(Message).where(
Message.id == number)
result = session.execute(statement).one()
self.api.print_message(result)
except sqlalchemy.exc.NoResultFound:
self.api.write_output(f"Message not found.")
except Exception as e:
print(e)
def run(self, args) -> None:
"""Read a message.
Arguments:
number -- the message number to read
"""
if args.number:
try:
result = self.api.controller.read(args)
self.api.print_message(result)
except sqlalchemy.exc.NoResultFound:
self.api.write_output(f"Message not found.")
except Exception as e:
print(e)
self.read_message(args.number)

21
rsbbs/plugins/readm/plugin.py
View File

@ -19,28 +19,41 @@
import sqlalchemy
from rsbbs.console import Console
from rsbbs.models import Message
from rsbbs.parser import Parser
class Plugin():
def __init__(self, api: Console):
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
if api.config.debug:
print(f"Plugin {__name__} loaded")
def init_parser(self, parser: Parser):
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='readm',
aliases=['rm'],
help='Read all messages addressed to you')
subparser.set_defaults(func=self.run)
def run(self, args):
def list_mine(self, args) -> sqlalchemy.ChunkedIteratorResult:
with self.api.controller.session() as session:
try:
statement = sqlalchemy.select(Message).where(
Message.recipient == self.api.config.calling_station)
result = session.execute(
statement,
execution_options={"prebuffer_rows": True})
return result
except Exception:
raise
def run(self, args) -> None:
"""Read all messages addressed to the calling station's callsign,
in sequence."""
result = self.api.controller.list_mine(args)
result = self.list_mine(args)
messages = result.all()
count = len(messages)
if count > 0:

26
rsbbs/plugins/send/plugin.py
View File

@ -16,21 +16,20 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import sqlalchemy
from rsbbs.console import Console
from rsbbs.models import Message
from rsbbs.parser import Parser
class Plugin():
def __init__(self, api: Console):
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
if api.config.debug:
print(f"Plugin {__name__} loaded")
def init_parser(self, parser: Parser):
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='send',
aliases=['s'],
@ -40,7 +39,22 @@ class Plugin():
subparser.add_argument('--message', help='Message')
subparser.set_defaults(func=self.run)
def run(self, args):
def send(self, args, is_private=False) -> None:
with self.api.controller.session() as session:
try:
session.add(Message(
sender=self.api.config.calling_station.upper(),
recipient=args.callsign.upper(),
subject=args.subject,
message=args.message,
is_private=is_private
))
session.commit()
except Exception:
session.rollback()
raise
def run(self, args) -> None:
"""Create a new message addressed to another callsign.
Required arguments:
@ -58,6 +72,6 @@ class Plugin():
args.message = self.api.read_multiline(
"Message - end with /ex on a single line:")
try:
self.api.controller.send(args, is_private=False)
self.send(args)
except Exception as e:
print(e)

26
rsbbs/plugins/sendp/plugin.py
View File

@ -16,21 +16,20 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
import sqlalchemy
from rsbbs.console import Console
from rsbbs.models import Message
from rsbbs.parser import Parser
class Plugin():
def __init__(self, api: Console):
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
if api.config.debug:
print(f"Plugin {__name__} loaded")
def init_parser(self, parser: Parser):
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='sendp',
aliases=['sp'],
@ -40,7 +39,22 @@ class Plugin():
subparser.add_argument('--message', help='Message')
subparser.set_defaults(func=self.run)
def run(self, args):
def send(self, args, is_private=False) -> None:
with self.api.controller.session() as session:
try:
session.add(Message(
sender=self.api.config.calling_station.upper(),
recipient=args.callsign.upper(),
subject=args.subject,
message=args.message,
is_private=is_private
))
session.commit()
except Exception:
session.rollback()
raise
def run(self, args) -> None:
"""Create a new message addressed to another callsign.
Required arguments:
@ -58,6 +72,6 @@ class Plugin():
args.message = self.api.read_multiline(
"Message - end with /ex on a single line:")
try:
self.api.controller.send(args, is_private=True)
self.send(args, is_private=True)
except Exception as e:
print(e)

1
rsbbs/rsbbs.py
View File

@ -23,7 +23,6 @@ from rsbbs import __version__
from rsbbs.config import Config
from rsbbs.console import Console
from rsbbs.controller import Controller
from rsbbs.parser import Parser
def parse_args():