Move models

This commit is contained in:
Minecon724 2025-04-02 18:57:30 +02:00
commit c86b0670a3
Signed by untrusted user who does not match committer: m724
GPG key ID: A02E6E67AB961189
7 changed files with 99 additions and 93 deletions

View file

@ -79,8 +79,6 @@ scheduler = APScheduler()
scheduler.init_app(app)
# Import models
from . import models
# Explicitly import all models to ensure they are registered with SQLAlchemy
from .models import Inquiry, Message, Settings, Admin
# Ensure tables are created when app is loaded by Gunicorn

View file

@ -1,91 +0,0 @@
from . import db
import os
import secrets
import argon2
from argon2 import PasswordHasher
from . import password_hasher
from datetime import datetime, timedelta
class Message(db.Model):
inquiry_id = db.Column(db.String(16), db.ForeignKey('inquiry.id'), nullable=False, primary_key=True)
message_number = db.Column(db.Integer, primary_key=True)
content = db.Column(db.Text, nullable=False)
timestamp = db.Column(db.DateTime, default=db.func.current_timestamp())
is_admin = db.Column(db.Boolean, default=False)
__table_args__ = (db.UniqueConstraint('inquiry_id', 'message_number'),)
def __init__(self, **kwargs):
super(Message, self).__init__(**kwargs)
if self.message_number is None:
# Find the highest message number for this inquiry and increment
last_message = Message.query.filter_by(
inquiry_id=self.inquiry_id
).order_by(Message.message_number.desc()).first()
self.message_number = 1 if last_message is None else last_message.message_number + 1
class Inquiry(db.Model):
id = db.Column(db.String(16), primary_key=True, unique=True, default=lambda: secrets.token_hex(8))
messages = db.relationship('Message', backref='inquiry', lazy=True,
cascade='all, delete-orphan',
order_by='Message.message_number')
is_closed = db.Column(db.Boolean, default=False)
closing_timestamp = db.Column(db.DateTime, nullable=True)
def close(self):
"""Mark an inquiry as closed"""
self.is_closed = True
self.closing_timestamp = datetime.utcnow()
def reopen(self):
"""Reopen a closed inquiry"""
self.is_closed = False
self.closing_timestamp = None
@staticmethod
def get_expired_inquiries(days=2):
"""Get inquiries that have been closed for more than the specified days"""
expiry_date = datetime.utcnow() - timedelta(days=days)
return Inquiry.query.filter(
Inquiry.is_closed == True,
Inquiry.closing_timestamp <= expiry_date
).all()
def get_deletion_date(self):
"""Get the deletion date for a closed inquiry"""
if self.is_closed:
return self.closing_timestamp + timedelta(days=2)
return None
class Settings(db.Model):
id = db.Column(db.Integer, primary_key=True)
webhook_enabled = db.Column(db.Boolean, default=False)
webhook_url = db.Column(db.String(255), nullable=True)
webhook_secret = db.Column(db.String(255), nullable=True)
class Admin(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), unique=True, nullable=False)
password_hash = db.Column(db.String(255), nullable=False)
@classmethod
def hash_password(cls, password):
"""Hash a password using Argon2id"""
return password_hasher.hash(password)
def rehash_password(self, password):
"""Rehash a password using Argon2id"""
self.password_hash = self.hash_password(password)
db.session.add(self)
db.session.commit()
def verify_password(self, password):
"""Verify a password against the stored hash"""
try:
password_hasher.verify(self.password_hash, password)
if password_hasher.check_needs_rehash(self.password_hash):
self.rehash_password(password)
return True
except argon2.exceptions.VerifyMismatchError:
return False

View file

@ -0,0 +1,6 @@
from .message import Message
from .inquiry import Inquiry
from .settings import Settings
from .admin import Admin
__all__ = ['Message', 'Inquiry', 'Settings', 'Admin']

View file

@ -0,0 +1,29 @@
from .. import db
import argon2
from .. import password_hasher
class Admin(db.Model):
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(64), unique=True, nullable=False)
password_hash = db.Column(db.String(255), nullable=False)
@classmethod
def hash_password(cls, password):
"""Hash a password using Argon2id"""
return password_hasher.hash(password)
def rehash_password(self, password):
"""Rehash a password using Argon2id"""
self.password_hash = self.hash_password(password)
db.session.add(self)
db.session.commit()
def verify_password(self, password):
"""Verify a password against the stored hash"""
try:
password_hasher.verify(self.password_hash, password)
if password_hasher.check_needs_rehash(self.password_hash):
self.rehash_password(password)
return True
except argon2.exceptions.VerifyMismatchError:
return False

View file

@ -0,0 +1,36 @@
from .. import db
import secrets
from datetime import datetime, timedelta
class Inquiry(db.Model):
id = db.Column(db.String(16), primary_key=True, unique=True, default=lambda: secrets.token_hex(8))
messages = db.relationship('Message', backref='inquiry', lazy=True,
cascade='all, delete-orphan',
order_by='Message.message_number')
is_closed = db.Column(db.Boolean, default=False)
closing_timestamp = db.Column(db.DateTime, nullable=True)
def close(self):
"""Mark an inquiry as closed"""
self.is_closed = True
self.closing_timestamp = datetime.utcnow()
def reopen(self):
"""Reopen a closed inquiry"""
self.is_closed = False
self.closing_timestamp = None
@staticmethod
def get_expired_inquiries(days=2):
"""Get inquiries that have been closed for more than the specified days"""
expiry_date = datetime.utcnow() - timedelta(days=days)
return Inquiry.query.filter(
Inquiry.is_closed == True,
Inquiry.closing_timestamp <= expiry_date
).all()
def get_deletion_date(self):
"""Get the deletion date for a closed inquiry"""
if self.is_closed:
return self.closing_timestamp + timedelta(days=2)
return None

View file

@ -0,0 +1,21 @@
from .. import db
from datetime import datetime
class Message(db.Model):
inquiry_id = db.Column(db.String(16), db.ForeignKey('inquiry.id'), nullable=False, primary_key=True)
message_number = db.Column(db.Integer, primary_key=True)
content = db.Column(db.Text, nullable=False)
timestamp = db.Column(db.DateTime, default=db.func.current_timestamp())
is_admin = db.Column(db.Boolean, default=False)
__table_args__ = (db.UniqueConstraint('inquiry_id', 'message_number'),)
def __init__(self, **kwargs):
super(Message, self).__init__(**kwargs)
if self.message_number is None:
# Find the highest message number for this inquiry and increment
last_message = Message.query.filter_by(
inquiry_id=self.inquiry_id
).order_by(Message.message_number.desc()).first()
self.message_number = 1 if last_message is None else last_message.message_number + 1

View file

@ -0,0 +1,7 @@
from .. import db
class Settings(db.Model):
id = db.Column(db.Integer, primary_key=True)
webhook_enabled = db.Column(db.Boolean, default=False)
webhook_url = db.Column(db.String(255), nullable=True)
webhook_secret = db.Column(db.String(255), nullable=True)