1
Fork 0
mirror of https://github.com/RYGhub/royalnet.git synced 2024-11-23 19:44:20 +00:00
This commit is contained in:
Steffo 2019-06-01 19:14:13 +02:00
parent 860172916e
commit 6eb2e35f45
6 changed files with 45 additions and 25 deletions

View file

@ -9,4 +9,7 @@ class TestConfig:
app = create_app(TestConfig, [helloworld.bp]) app = create_app(TestConfig, [helloworld.bp])
if __name__ == "__main__":
app.run() app.run()

View file

@ -1,3 +1,4 @@
from .flaskserver import create_app from .flaskserver import create_app
from .royalprint import Royalprint
__all__ = ["create_app"] __all__ = ["create_app", "Royalprint"]

View file

@ -1,18 +0,0 @@
import flask as f
import werkzeug.local
def get_alchemy_session():
if "alchemy_session" not in f.g:
f.g.alchemy_session = f.g.alchemy.Session()
return f.g.alchemy_session
@f.current_app.teardown_appcontext
def teardown_alchemy_session(*_, **__):
_alchemy_session = f.g.pop("alchemy_session", None)
if _alchemy_session is not None:
_alchemy_session.close()
alchemy_session = werkzeug.local.LocalProxy(get_alchemy_session)

View file

@ -1,9 +1,24 @@
import flask as f import flask as f
from .. import Royalprint
from ...database.tables import Royal
bp = f.Blueprint("helloworld", __name__, url_prefix="/helloworld") bp = Royalprint("helloworld", __name__, url_prefix="/helloworld", required_tables={Royal})
@bp.route("/") @bp.route("/")
def helloworld(): def helloworld():
return "Hello world!" royals = f.g.alchemy_session.query(f.g.alchemy.Royal).all()
return repr(royals)
@bp.before_request
def before_request():
f.g.alchemy_session = f.g.alchemy.Session()
@bp.after_request
def after_request():
alchemy_session = f.g.pop("alchemy_session", None)
if alchemy_session is not None:
alchemy_session.close()

View file

@ -1,13 +1,16 @@
import typing import typing
import flask as f import flask as f
from ..database import Alchemy from ..database import Alchemy
from .royalprint import Royalprint
def create_app(config_obj: typing.Type, blueprints: typing.List[f.Blueprint]): def create_app(config_obj: typing.Type, blueprints: typing.List[Royalprint]):
app = f.Flask(__name__) app = f.Flask(__name__)
app.config.from_object(config_obj) app.config.from_object(config_obj)
with app.app_context(): required_tables = set()
f.g.alchemy = Alchemy(app.config["DB_PATH"], app.config["REQUIRED_TABLES"])
for blueprint in blueprints: for blueprint in blueprints:
required_tables.union(blueprint.required_tables)
app.register_blueprint(blueprint) app.register_blueprint(blueprint)
with app.app_context():
f.g.alchemy = Alchemy(app.config["DB_PATH"], required_tables)
return app return app

View file

@ -0,0 +1,16 @@
import typing
import flask as f
class Royalprint(f.Blueprint):
def __init__(self, name, import_name, static_folder=None,
static_url_path=None, template_folder=None,
url_prefix=None, subdomain=None, url_defaults=None,
root_path=None, required_tables: typing.Optional[set] = None):
super().__init__(name, import_name, static_folder=static_folder,
static_url_path=static_url_path, template_folder=template_folder,
url_prefix=url_prefix, subdomain=subdomain, url_defaults=url_defaults,
root_path=root_path)
self.required_tables = required_tables
if self.required_tables is None:
self.required_tables = set()