mirror of
https://github.com/nelsbrock/NateMan.git
synced 2026-08-14 21:36:49 +02:00
270 lines
10 KiB
Python
270 lines
10 KiB
Python
# NateMan – Nachschreibtermin-Manager
|
||
# models.py
|
||
# Copyright © 2020 Niklas Elsbrock
|
||
#
|
||
# 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/>.
|
||
|
||
"""
|
||
Enthält die SQLAlchemy-Modelle.
|
||
"""
|
||
from datetime import datetime, timedelta
|
||
from sqlite3 import Connection as Sqlite3_Connection
|
||
from typing import Optional, Tuple, Union, AnyStr
|
||
|
||
import bcrypt
|
||
from flask_sqlalchemy import SQLAlchemy
|
||
from sqlalchemy import CheckConstraint
|
||
from sqlalchemy.engine import Engine
|
||
from sqlalchemy.sql import func
|
||
|
||
from . import util
|
||
from .config_manager import config
|
||
|
||
db: SQLAlchemy = SQLAlchemy()
|
||
|
||
|
||
class Klausurteilnahme(db.Model):
|
||
klausur_id = db.Column(db.Integer, db.ForeignKey("klausur.id", onupdate="CASCADE", ondelete="CASCADE"),
|
||
primary_key=True)
|
||
schueler_id = db.Column(db.Integer, db.ForeignKey("schueler.id", onupdate="CASCADE", ondelete="CASCADE"),
|
||
primary_key=True)
|
||
versaeumt = db.Column(db.Boolean, default=False, nullable=False)
|
||
attestiert = db.Column(db.Boolean, default=False, nullable=False)
|
||
nachgeschrieben = db.Column(db.Boolean, default=False, nullable=False)
|
||
|
||
klausur = db.relationship("Klausur", lazy="joined")
|
||
schueler = db.relationship("Schueler", lazy="joined")
|
||
|
||
|
||
class Stufe(db.Model):
|
||
name = db.Column(db.String(), primary_key=True)
|
||
import_date = db.Column(db.Date(), default=None)
|
||
|
||
|
||
class Schueler(db.Model):
|
||
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
|
||
nachname = db.Column(db.String(collation="NOCASE"), nullable=False)
|
||
vorname = db.Column(db.String(collation="NOCASE"), nullable=False)
|
||
stufe_name = db.Column(db.String(), db.ForeignKey("stufe.name", onupdate="CASCADE", ondelete="CASCADE"),
|
||
nullable=False)
|
||
stammschule_name = db.Column(db.String(),
|
||
db.ForeignKey("koopschule.kuerzel", onupdate="CASCADE", ondelete="CASCADE"))
|
||
|
||
stufe = db.relationship("Stufe", lazy="select")
|
||
stammschule = db.relationship("Koopschule", lazy="select")
|
||
|
||
def __str__(self):
|
||
return f"{self.nachname}, {self.vorname} ({self.stufe.name}, ID:{self.id})"
|
||
|
||
|
||
class Lehrer(db.Model):
|
||
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
|
||
kuerzel = db.Column(db.String(collation="NOCASE"), unique=True, nullable=False)
|
||
email = db.Column(db.String())
|
||
is_confirmed = db.Column(db.Boolean, default=False, nullable=False)
|
||
pwd_hash = db.Column(db.String(), nullable=False)
|
||
is_admin = db.Column(db.Boolean, default=False, nullable=False)
|
||
beraet_name = db.Column(db.String(), db.ForeignKey("stufe.name", onupdate="CASCADE", ondelete="SET NULL"))
|
||
pwd_changed = db.Column(db.Boolean, default=False, nullable=False)
|
||
confirmation_token = db.Column(db.String(), unique=True)
|
||
password_reset_token = db.Column(db.String(), unique=True)
|
||
password_reset_expiry = db.Column(db.DateTime)
|
||
|
||
beraet = db.relationship("Stufe", lazy="select")
|
||
|
||
@staticmethod
|
||
def password_reset_cleanup() -> None:
|
||
"""
|
||
Entfernt alle abgelaufenen Passwortzurücksetzungsanfragen aus der Datenbank.
|
||
"""
|
||
Lehrer.query.filter(Lehrer.password_reset_expiry < datetime.utcnow()) \
|
||
.update({Lehrer.password_reset_token: None, Lehrer.password_reset_expiry: None})
|
||
|
||
def __str__(self):
|
||
return f"{self.kuerzel} (ID:{self.id})"
|
||
|
||
def can_access(self, stufe: Optional[Stufe] = None) -> bool:
|
||
"""
|
||
Überprüft, ob dieser Lehrer erweiterten Zugriff auf die angegebene Stufe hat
|
||
(d.h. ob er Administrator oder Beratungslehrer der angegebenen Stufe ist).
|
||
|
||
Wenn die Stufe ``None`` ist, wird überprüft, ob dieser Lehrer
|
||
Administrator oder Beratungslehrer irgendeiner Stufe ist.
|
||
|
||
:return: ``True`` falls ja, ``False`` falls nicht
|
||
"""
|
||
if self.is_admin:
|
||
return True
|
||
# ELSE
|
||
if stufe is None:
|
||
return self.beraet is not None
|
||
# ELSE
|
||
return self.beraet == stufe
|
||
|
||
def accessible_stufen(self, names: bool = False) -> Union[Tuple[Stufe, ...], Tuple[AnyStr, ...]]:
|
||
"""
|
||
:param names: Falls ``True``, werden die Namen der Stufen anstelle der Stufenobjekte zurückgegeben
|
||
:return: Tupel mit allen Stufen, auf die dieser Lehrer erweiterten Zugriff hat
|
||
"""
|
||
stufen = Stufe.query.all()
|
||
if self.is_admin:
|
||
return tuple(stufen) if not names else tuple(s.name for s in stufen)
|
||
# ELSE
|
||
if self.beraet:
|
||
return (self.beraet,) if not names else (self.beraet.name,)
|
||
# ELSE
|
||
return ()
|
||
|
||
def set_password(self, new_password: str, set_pwd_changed=True) -> None:
|
||
"""
|
||
Legt das neue Passwort des Lehrers fest.
|
||
|
||
:param new_password: neues Passwort
|
||
:param set_pwd_changed: legt fest, ob der ``pwd_changed``-Eintrag in der Datenbank angepasst werden soll
|
||
"""
|
||
assert util.validate_bcrypt_password(new_password)
|
||
self.pwd_hash = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt())
|
||
if set_pwd_changed:
|
||
self.pwd_changed = True
|
||
|
||
def check_password(self, password) -> bool:
|
||
"""
|
||
Vergleicht ``password`` mit dem Passwort des Lehrers.
|
||
|
||
:param password: zu überprüfendes Passwort
|
||
:return: ``True`` falls die Passwörter übereinstimmen, ``False`` falls nicht
|
||
"""
|
||
return bcrypt.checkpw(password.encode("utf-8"), self.pwd_hash)
|
||
|
||
def set_default_email_address(self, fmt) -> bool:
|
||
email_address = fmt.replace("?", util.email_address_safe(self.kuerzel))
|
||
if util.EMAIL_ADDRESS_REGEX.match(email_address):
|
||
self.email = email_address
|
||
self.is_confirmed = True
|
||
return True
|
||
|
||
return False
|
||
|
||
|
||
class Klausur(db.Model):
|
||
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
|
||
kursname = db.Column(db.String(collation="NOCASE"), nullable=False)
|
||
date = db.Column(db.Date, nullable=False)
|
||
startperiod = db.Column(db.Integer, nullable=False)
|
||
endperiod = db.Column(db.Integer, nullable=False)
|
||
laenge = db.Column(db.Integer, default=None)
|
||
edited = db.Column(db.Boolean, default=False, nullable=False)
|
||
annotation = db.Column(db.String())
|
||
stufe_name = db.Column(db.String(), db.ForeignKey("stufe.name", onupdate="CASCADE", ondelete="CASCADE"),
|
||
nullable=False)
|
||
lehrer_id = db.Column(db.Integer, db.ForeignKey("lehrer.id", onupdate="CASCADE", ondelete="CASCADE"),
|
||
nullable=False)
|
||
|
||
stufe = db.relationship("Stufe", lazy="select")
|
||
lehrer = db.relationship("Lehrer", lazy="joined")
|
||
|
||
__table_args__ = (
|
||
CheckConstraint("startperiod >= 0"),
|
||
CheckConstraint("startperiod <= endperiod")
|
||
)
|
||
|
||
def __str__(self):
|
||
return f"{self.kursname}, {self.stufe.name} (ID:{self.id})"
|
||
|
||
def is_bygone(self):
|
||
"""
|
||
:return: ``True``, falls diese Klausur vergangen ist, ``False``, falls nicht
|
||
"""
|
||
return self.date <= datetime.now().date()
|
||
|
||
|
||
class Koopschule(db.Model):
|
||
kuerzel = db.Column(db.String(), primary_key=True)
|
||
name = db.Column(db.String())
|
||
|
||
|
||
class Session(db.Model):
|
||
key = db.Column(db.String(), primary_key=True)
|
||
expiry = db.Column(db.DateTime, nullable=False)
|
||
lehrer_id = db.Column(db.Integer, db.ForeignKey("lehrer.id", onupdate="CASCADE", ondelete="CASCADE"),
|
||
nullable=False)
|
||
|
||
lehrer = db.relationship("Lehrer", lazy="joined")
|
||
|
||
@staticmethod
|
||
def get(key: str) -> Optional["Session"]:
|
||
"""
|
||
Gibt die Sitzung zurück, die zum Key gehört oder ``None``,
|
||
falls keine Sitzung mit dem Key existiert oder die Sitzung abgelaufen ist.
|
||
|
||
:param key: Sitzungs-Key
|
||
:return: Sitzung
|
||
"""
|
||
session = Session.query.filter_by(key=key).first()
|
||
if session is not None and session.expired():
|
||
return None
|
||
|
||
return session
|
||
|
||
@staticmethod
|
||
def create(lehrer: Lehrer, expiry_delta: timedelta) -> "Session":
|
||
"""
|
||
Erstellt eine neue Sitzung für den Lehrer ``lehrer``.
|
||
|
||
:param lehrer: Lehrer, für den eine Sitzung erstellt werden soll
|
||
:param expiry_delta: Zeitraum, nach dem die Sitzung ablaufen soll
|
||
:return: neue Sitzung
|
||
"""
|
||
new_session = Session(key=util.random_uri_safe_string(64), expiry=datetime.utcnow() + expiry_delta,
|
||
lehrer=lehrer)
|
||
db.session.add(new_session)
|
||
return new_session
|
||
|
||
@staticmethod
|
||
def cleanup() -> None:
|
||
"""
|
||
Entfernt alle abgelaufenen Sitzungen aus der Datenbank.
|
||
"""
|
||
Session.query.filter(Session.expiry < datetime.utcnow()).delete()
|
||
|
||
def expired(self) -> bool:
|
||
"""
|
||
Überprüft, ob die Sitzung abgelaufen ist.
|
||
"""
|
||
return self.expiry < datetime.utcnow()
|
||
|
||
|
||
@db.event.listens_for(Engine, "connect")
|
||
def set_sqlite_pragma(dbapi_connection, connection_record):
|
||
"""
|
||
Falls SQLite verwendet wird, wird bei jedem Verbindungsaufbau zur Datenbank ``foreign_keys`` aktiviert, damit
|
||
Fremdschlüssel richtig funktionieren.
|
||
Von https://stackoverflow.com/questions/4477269.
|
||
"""
|
||
if type(dbapi_connection) is Sqlite3_Connection:
|
||
cursor = dbapi_connection.cursor()
|
||
cursor.execute("PRAGMA foreign_keys=ON")
|
||
cursor.close()
|
||
|
||
|
||
def get_next_new_schueler_id():
|
||
"""
|
||
:return: ID, die der nächste manuell hinzugefügte Schüler haben soll
|
||
"""
|
||
lowest_id = db.session.query(func.min(Schueler.id).label("min_id")).one().min_id
|
||
if lowest_id is None or lowest_id > -1:
|
||
return -1
|
||
# ELSE
|
||
return lowest_id - 1
|