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

move session to root class; add support for marking messages 'read'

Browse Source
This commit is contained in:
John Burwell 2023-04-30 19:40:37 -05:00
parent c8226492a4
commit 98d8941152
6 changed files with 158 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
rsbbs/controller.py
View File

@ -28,6 +28,7 @@ class Controller():
def __init__(self, config: Config) -> None:
self.config = config
self._init_datastore()
self._session
def _init_datastore(self) -> None:
"""Create a connection to the sqlite3 database.
@ -42,5 +43,7 @@ class Controller():
# Create the database schema if none exists
Base.metadata.create_all(self.engine)
self._session = Session(self.engine, autoflush=True)
def session(self) -> Session:
return Session(self.engine)
return self._session

20
rsbbs/models.py
View File

@ -18,16 +18,27 @@
from datetime import datetime, timezone
from sqlalchemy import Boolean, DateTime, String, Integer
from sqlalchemy import Boolean, DateTime, String, Integer,\
Table, ForeignKey, Column
from sqlalchemy.orm import DeclarativeBase, Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import mapped_column, relationship, backref
class Base(DeclarativeBase):
pass
# Define the association table that links users and messages
user_message_table = Table('user_message', Base.metadata,
Column('user_id',
Integer, ForeignKey('user.id')),
Column('message_id',
Integer, ForeignKey('message.id')))
# Messages
class Message(Base):
__tablename__ = 'message'
id: Mapped[int] = mapped_column(primary_key=True)
@ -40,6 +51,8 @@ class Message(Base):
is_private: Mapped[bool] = mapped_column(Boolean)
# Users
class User(Base):
__tablename__ = 'user'
id: Mapped[int] = mapped_column(primary_key=True)
@ -49,3 +62,6 @@ class User(Base):
login_count: Mapped[int] = mapped_column(Integer)
login_last: Mapped[DateTime] = mapped_column(
DateTime, default=datetime.now(timezone.utc))
messages = relationship('Message',
secondary=user_message_table,
backref='read_by')

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

@ -34,14 +34,15 @@ class Plugin():
subparser = parser.subparsers.add_parser(
name='listm',
aliases=['lm'],
help='List only messages addressed to you')
help='List 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:
callsign = self.api.config.calling_station
statement = sqlalchemy.select(Message).where(
Message.recipient == self.api.config.calling_station)
Message.recipient == callsign)
result = session.execute(
statement,
execution_options={"prebuffer_rows": True})

61
rsbbs/plugins/listnew/plugin.py Normal file
View File

@ -0,0 +1,61 @@
#!/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/>.
import logging
import sqlalchemy
from rsbbs import Console, Parser
from rsbbs.models import Message, User
class Plugin():
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
logging.info(f"plugin {__name__} loaded")
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='listm',
aliases=['ln'],
help='List unread 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:
callsign = self.api.config.calling_station
statement = sqlalchemy.select(Message).where(
Message.recipient == callsign).where(
~Message.read_by.any(User.id == self.api.user.id)
)
result = session.execute(
statement,
execution_options={"prebuffer_rows": True})
logging.info("list my messages")
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.list_mine(args)
self.api.print_message_list(result)

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

@ -21,7 +21,7 @@ import sqlalchemy
import sqlalchemy.exc
from rsbbs import Console, Parser
from rsbbs.models import Message
from rsbbs.models import Message, User
class Plugin():
@ -47,6 +47,11 @@ class Plugin():
result = session.execute(statement).one()
self.api.print_message(result)
logging.info(f"read message")
session.commit()
user = session.get(User, self.api.user.id)
user.messages.append(result[0])
logging.info(f"User {user.id} read message {result[0].id}")
session.commit()
except sqlalchemy.exc.NoResultFound:
self.api.write_output(f"Message not found.")
except Exception as e:

66
rsbbs/plugins/readnew/plugin.py Normal file
View File

@ -0,0 +1,66 @@
#!/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/>.
import logging
import sqlalchemy
from rsbbs import Console, Parser
from rsbbs.models import Message, User
class Plugin():
def __init__(self, api: Console) -> None:
self.api = api
self.init_parser(api.parser)
logging.info(f"plugin {__name__} loaded")
def init_parser(self, parser: Parser) -> None:
subparser = parser.subparsers.add_parser(
name='readnew',
aliases=['rn'],
help='Read all unread 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
and self.api.user.id not in Message.read_by)
result = session.execute(
statement,
execution_options={"prebuffer_rows": True})
logging.info(f"read message")
return result
except Exception:
raise
def run(self, args) -> None:
"""Read all messages addressed to the calling station's callsign,
in sequence."""
result = self.list_mine(args)
messages = result.all()
count = len(messages)
if count > 0:
self.api.write_output(f"Reading {count} messages:")
for message in messages:
self.api.print_message(message)
self.api.read_enter("Enter to continue...")
else:
self.api.write_output(f"No messages to read.")