```
$ FLASK_APP=wradmin FLASK_DEBUG=1 WRADMIN_SETTINGS=development.cfg ipython
->>> import wradmin.model
->>> wradmin.model.meta.Session.query(wradmin.model.MwUser).get(1)
+>>> from wradmin.app import db
+>>> from wradmin.model import MwUser
+>>> user = db.session.query(MwUser).get(1)
```
## Deploy
Development environment:
-```
+```{sh}
$ python setup.py bdist_wheel
$ scp bdist/wradmin-2.1.0.dev0-py3-none-any.whl colleen:
```
Server environment:
-```
+```{sh}
$ bin/wradmin stop
$ pip uninstall wradmin
$ pip install wradmin-2.1.0.dev0-py3-none-any.whl # or similar
"Flask-Login",
"Flask-Principal",
"Flask-WTF",
+ "Flask-SQLAlchemy",
"mysqlclient",
],
packages=['wradmin', 'wradmin.auth', 'wradmin.model', 'wradmin.controllers'],
import unittest
+
import bs4 # beautiful soup
+
+import wradmin.app
import wradmin.model
-import wradmin
+from wradmin.app import db
class WradminTestBase(unittest.TestCase):
def setUp(self):
wradmin.app.config['TESTING'] = True
self.app = wradmin.app.test_client()
+ self.db = db
def login(self):
return self.app.post('/login', data={'user_name': 'Johndoe', 'password': 'doejohn'}, follow_redirects=True)
# fill database
with open('tests/testdb.sql', 'r') as f:
sql = f.read()
- with wradmin.model.meta.engine.begin() as con:
+ with wradmin.db.engine.begin() as con:
con.execution_options(no_parameters=True).execute(sql)
# update dates
self.assertIn('2018-01-30 15:09:00', str(soup))
def test_gasthaus_list(self):
-
result = self.login_and_get('/gasthaus/list')
self.assertEqual(result.status_code, 200)
self.assertTrue(result.data.startswith(b'<!doctype html'))
self.assertIn('Die Gasthausliste wurde erfolgreich aktualisiert.', str(soup))
def test_gasthaus_update_fail(self):
- session = wradmin.model.meta.Session
- text = session.query(wradmin.model.MwText).get(14415)
+ text = self.db.session.query(wradmin.model.MwText).get(14415)
text.old_text = text.old_text.replace('lindauerhuette@aon.at', 'abc@def@example.com')
- session.commit()
+ self.db.session.commit()
self.login()
result = self.app.get('/gasthaus/update', follow_redirects=True)
self.assertEqual(result.status_code, 200)
self.assertIn('Die Rodelbahnliste wurde erfolgreich aktualisiert.', str(soup))
def test_rodelbahn_update_fail(self):
- session = wradmin.model.meta.Session
- text = session.query(wradmin.model.MwText).get(15032)
+ text = self.db.session.query(wradmin.model.MwText).get(15032)
text.old_text = text.old_text.replace('Schwierigkeit = mittel', 'Schwierigkeit = geht so')
- session.commit()
+ self.db.session.commit()
self.login()
result = self.app.get('/rodelbahn/update', follow_redirects=True)
self.assertEqual(result.status_code, 200)
from flask import Flask, send_from_directory, abort, g, render_template, request, redirect, url_for, flash, \
session, current_app
+from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.engine import create_engine
import wradmin.model
import wradmin.template_helper
+from wradmin.app import app, db, admin_permission, login_manager
from wradmin.controllers.rodelbahn import RodelbahnController
from wradmin.controllers.gasthaus import GasthausController
from wradmin.controllers.bericht import BerichtController
AnonymousIdentity, UserNeed
-app = Flask(__name__)
-app.config.from_envvar('WRADMIN_SETTINGS')
-wradmin.model.init_model(create_engine(app.config['DATABASE_URI']))
-app.jinja_env.globals.update(h=wradmin.template_helper.PylonsHelper())
-login_manager = LoginManager(app)
-login_manager.login_view = "login"
-principals = Principal(app)
-admin_permission = Permission(RoleNeed('admin'))
-
-
@app.before_request
def _before_request():
g.user = current_user
-@app.teardown_appcontext
-def remove_db_session(error):
- """Removes the database session at the end of the request."""
- wradmin.model.meta.Session.remove()
-
-
@app.route("/")
def index():
return render_template('index.html')
def login():
form = LoginForm()
if form.validate_on_submit():
- user = wradmin.model.meta.Session.query(wradmin.model.MwUser).filter_by(user_name=form.user_name.data).first()
+ user = db.session.query(wradmin.model.MwUser).filter_by(user_name=form.user_name.data).first()
if user is not None and password_is_correct(form.password.data, user.user_password.decode()):
login_user(user, form.remember_me.data)
identity_changed.send(current_app._get_current_object(), identity=Identity(user.get_id()))
@login_manager.user_loader
def user_loader(user_id):
- return wradmin.model.meta.Session.query(wradmin.model.MwUser).get(user_id)
+ return wradmin.db.session.query(wradmin.model.MwUser).get(user_id)
@identity_loaded.connect_via(app)
user_id = current_user.get_id()
if user_id is not None:
identity.provides.add(UserNeed(user_id))
- if wradmin.model.meta.Session.query(wradmin.model.MwUserGroups).get((user_id, 'sysop')) is not None:
+ if wradmin.db.session.query(wradmin.model.MwUserGroups).get((user_id, 'sysop')) is not None:
identity.provides.add(RoleNeed('admin'))
--- /dev/null
+from flask import Flask
+from flask_login import LoginManager
+from flask_principal import Principal, Permission, RoleNeed
+
+import wradmin
+
+app = Flask(__name__)
+app.config.from_envvar('WRADMIN_SETTINGS')
+db = wradmin.model.init_db(app)
+app.jinja_env.globals.update(h=wradmin.template_helper.PylonsHelper())
+login_manager = LoginManager(app)
+login_manager.login_view = "login"
+principals = Principal(app)
+admin_permission = Permission(RoleNeed('admin'))
from flask import request, abort, redirect, url_for, flash, render_template
import paginate
import sqlalchemy as sa
-import wradmin.model as model
+
import wrpylib.wrmwcache
+from wradmin import db
+from wradmin.model import WrReport
+
class BerichtController:
def list(self):
"""Lists all reports"""
- q = model.meta.Session.query(model.WrReport)
- q = q.order_by(sa.sql.expression.desc(model.WrReport.id))
+ q = db.session.query(WrReport)
+ q = q.order_by(sa.sql.expression.desc(WrReport.id))
paginator = paginate.Page(q.all(), page=int(request.args.get('page', 1)),
url_maker=lambda page: url_for('bericht_list', page=page),
items_per_page=25)
def view(self, id):
"""Displays a report"""
- q = model.meta.Session.query(model.WrReport)
+ q = db.session.query(WrReport)
wrreport = q.get(id)
if wrreport is None:
abort(404)
assert request.method == 'POST'
if id is None:
abort(404)
- wrreport_q = model.meta.Session.query(model.WrReport)
+ wrreport_q = db.session.query(WrReport)
wrreport = wrreport_q.get(id)
if wrreport is None:
abort(404)
flash(str(e))
return redirect(url_for('bericht_view', id=id))
wrreport.date_invalid = date_userdefined
- model.meta.Session.commit()
+ db.session.commit()
flash('Datum wurde erfolgreich geändert')
return redirect(url_for('bericht_view', id=id))
def update_reportcache(self):
"""Updates the wrreportcache table."""
- c = model.meta.Session.connection()
+ c = db.session.connection()
try:
wrpylib.wrmwcache.update_wrreportcache(c)
- model.meta.Session.commit()
+ db.session.commit()
flash('Die Berichtzusammenfassung-Tabelle wurde erfolgreich aktualisiesrt.', 'info')
except wrpylib.wrmwcache.UpdateCacheError as e:
flash(str(e), 'error')
import wrpylib.wrmwmarkup
import wrpylib.wrmwcache
import wrpylib.mwmarkup
-
-import wradmin.model as model
+from wradmin.app import db
+from wradmin.model import WrInnCache
class GasthausController:
def list(self):
"""Lists all inns"""
- q = model.meta.Session.query(model.WrInnCache)
- q = q.order_by(model.WrInnCache.page_title)
+ q = db.session.query(WrInnCache)
+ q = q.order_by(WrInnCache.page_title)
paginator = paginate.Page(q.all(), page=int(request.args.get('page', 1)),
url_maker=lambda page: url_for('gasthaus_list', page=page),
items_per_page=25)
def view(self, id):
"""Displays an inn"""
- q = model.meta.Session.query(model.WrInnCache)
+ q = db.session.query(WrInnCache)
inn = q.get(id)
if inn is None:
abort(404)
def update(self):
"""Updates the wrinncache table from the wiki"""
- c = model.meta.Session.connection()
+ c = db.session.connection()
try:
wrpylib.wrmwcache.update_wrinncache(c)
- model.meta.Session.commit()
+ db.session.commit()
flash('Die Gasthausliste wurde erfolgreich aktualisiert.', 'info')
except wrpylib.wrmwcache.UpdateCacheError as e:
title = str(e.args[1])
import wrpylib.wrmwcache
import wrpylib.mwmarkup
-import wradmin.model as model
+from wradmin.app import db
+from wradmin.model import WrSledrunCache, WrReport
class RodelbahnController:
def list(self):
"""Lists all sled runs"""
- q = model.meta.Session.query(model.WrSledrunCache)
- q = q.order_by(model.WrSledrunCache.page_title)
+ q = db.session.query(WrSledrunCache)
+ q = q.order_by(WrSledrunCache.page_title)
paginator = paginate.Page(q.all(), page=int(request.args.get('page', 1)),
url_maker=lambda page: url_for('rodelbahn_list', page=page),
items_per_page=25)
def view(self, id):
"""Displays a sled run"""
- q = model.meta.Session.query(model.WrSledrunCache)
+ q = db.session.query(WrSledrunCache)
sledding = q.get(id)
if sledding is None:
abort(404)
- q = model.meta.Session.query(model.WrReport)
- q = q.filter_by(page_id=id).order_by(sa.sql.expression.desc(model.WrReport.id))
+ q = db.session.query(WrReport)
+ q = q.filter_by(page_id=id).order_by(sa.sql.expression.desc(WrReport.id))
paginator = paginate.Page(q.all(), page=int(request.args.get('page', 1)),
url_maker=lambda page: url_for('rodelbahn_view', id=id, page=page),
items_per_page=25)
def update(self):
"""Updates the wrsledruncache table from the wiki"""
- c = model.meta.Session.connection()
+ c = db.session.connection()
try:
wrpylib.wrmwcache.update_wrsledruncache(c)
- model.meta.Session.commit()
+ db.session.commit()
flash('Die Rodelbahnliste wurde erfolgreich aktualisiert.', 'info')
except wrpylib.wrmwcache.UpdateCacheError as e:
title = str(e.args[1])
def update_regioncache(self):
"""Updates the wrregioncache table from the wiki"""
- c = model.meta.Session.connection()
+ c = db.session.connection()
try:
wrpylib.wrmwcache.update_wrregioncache(c)
- model.meta.Session.commit()
+ db.session.commit()
flash('Die Rodelbahneinträge in den Regionslisten wurden erfolgreich aktualisiert.', 'info')
except wrpylib.wrmwcache.UpdateCacheError as e:
flash(str(e), 'error')
def update_mapcache(self):
"""Updates the wrmappointcache and wrmappathcache tables from the wiki."""
- c = model.meta.Session.connection()
+ c = db.session.connection()
try:
wrpylib.wrmwcache.update_wrmapcache(c)
- model.meta.Session.commit()
+ db.session.commit()
flash('Die Landkarteninformationen aus dem Wiki wurden erfolgreich aktualisiesrt.', 'info')
except wrpylib.wrmwcache.UpdateCacheError as e:
flash(str(e), 'error')
"""Model of wradmin"""
+from flask import Flask
+from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import orm
from flask_login import UserMixin
-from wradmin.model import meta
from wrpylib import mwdb, wrmwdb
-def init_model(engine):
- """Call me before using any of the tables or classes in the model"""
- meta.Session.configure(bind=engine)
- meta.engine = engine
-
-
-wrreport_table = wrmwdb.wrreport_table(meta.metadata)
-wrreportcache_table = wrmwdb.wrreportcache_table(meta.metadata)
-wrsledruncache_table = wrmwdb.wrsledruncache_table(meta.metadata)
-wrinncache_table = wrmwdb.wrinncache_table(meta.metadata)
-page_table = mwdb.page_table(meta.metadata)
-revision_table = mwdb.revision_table(meta.metadata)
-text_table = mwdb.text_table(meta.metadata)
-user_table = mwdb.user_table(meta.metadata)
-user_groups_table = mwdb.user_groups_table(meta.metadata)
-categorylinks_table = mwdb.categorylinks_table(meta.metadata)
-
-
-class WrReport(object):
- pass
-
-
-# Old version (not mapped)
-class WrSleddingCache1_2(object):
- pass
-
-
-class WrSledrunCache(object):
- pass
-
-
-# Old version (not mapped)
-class WrInnCache1_2(object):
+class WrReport:
pass
-class WrInnCache(object):
+class WrSledrunCache:
pass
-# Page (not mapped)
-class Page(object):
+class WrInnCache:
pass
-# MediaWiki text table
-class MwText(object):
+class MwText:
+ """MediaWiki text table"""
pass
-# MediaWiki user table
class MwUser(UserMixin):
+ """MediaWiki user table"""
def get_id(self):
return str(self.user_id)
-# MediaWiki groups
class MwUserGroups:
- pass
-
-
-orm.mapper(WrReport, wrreport_table)
-# We could add a relation but we don't need it yet:
-# orm.mapper(WrSledrunCache, wrsledruncache_table, properties = {'reports': orm.relation(WrReport, backref='sledding')})
-orm.mapper(WrSledrunCache, wrsledruncache_table)
-orm.mapper(WrInnCache, wrinncache_table)
-orm.mapper(MwText, text_table)
-orm.mapper(MwUserGroups, user_groups_table)
-orm.mapper(MwUser, user_table)
+ """MediaWiki groups"""
+ pass
+
+
+def init_db(app: Flask) -> SQLAlchemy:
+ db = SQLAlchemy(app)
+
+ wrreport_table = wrmwdb.wrreport_table(db.metadata)
+ wrreportcache_table = wrmwdb.wrreportcache_table(db.metadata)
+ wrsledruncache_table = wrmwdb.wrsledruncache_table(db.metadata)
+ wrinncache_table = wrmwdb.wrinncache_table(db.metadata)
+ page_table = mwdb.page_table(db.metadata)
+ revision_table = mwdb.revision_table(db.metadata)
+ text_table = mwdb.text_table(db.metadata)
+ user_table = mwdb.user_table(db.metadata)
+ user_groups_table = mwdb.user_groups_table(db.metadata)
+ categorylinks_table = mwdb.categorylinks_table(db.metadata)
+
+ orm.mapper(WrReport, wrreport_table)
+ # We could add a relation but we don't need it yet:
+ # orm.mapper(WrSledrunCache, wrsledruncache_table,
+ # properties = {'reports': orm.relation(WrReport, backref='sledding')})
+ orm.mapper(WrSledrunCache, wrsledruncache_table)
+ orm.mapper(WrInnCache, wrinncache_table)
+ orm.mapper(MwText, text_table)
+ orm.mapper(MwUserGroups, user_groups_table)
+ orm.mapper(MwUser, user_table)
+
+ return db
+++ /dev/null
-"""SQLAlchemy Metadata and Session object"""
-from sqlalchemy import MetaData
-from sqlalchemy.orm import scoped_session, sessionmaker
-
-__all__ = ['Session', 'engine', 'metadata']
-
-# SQLAlchemy database engine. Updated by model.init_model()
-engine = None
-
-# SQLAlchemy session manager. Updated by model.init_model()
-Session = scoped_session(sessionmaker(autoflush=True, autocommit=False))
-
-# Global metadata. If you have multiple databases with overlapping table
-# names, you'll need a metadata for each database
-metadata = MetaData()
DEBUG = False
SECRET_KEY = b'.\xe2{\xa4ns\xb7\xbb\x0cHR\xdf@}\xfa\x05;.K\x9bQ\xbc!='
-DATABASE_URI = 'mysql://philipp@localhost:3306/philipp_winterrodeln_wiki_wradmin_test?charset=utf8&use_unicode=1'
+SQLALCHEMY_DATABASE_URI = 'mysql://philipp@localhost:3306/philipp_winterrodeln_wiki_wradmin_test?charset=utf8mb4'
WTF_CSRF_ENABLED = False
+SQLALCHEMY_TRACK_MODIFICATIONS = False
ToastFreeware / philipp / winterrodeln / wradmin.git / commitdiff