Source code for eventsourcing.example.interface.flaskapp

import os

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy_utils.types.uuid import UUIDType

from eventsourcing.example.application import (
    get_example_application,
    init_example_application,
)
from eventsourcing.infrastructure.sqlalchemy.manager import SQLAlchemyRecordManager

# Read DB URI from environment.
uri = os.environ.get("DB_URI", "sqlite:///:memory:")


# Construct Flask application.
application = Flask(__name__)
application.config["SQLALCHEMY_DATABASE_URI"] = uri
application.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

# Define database connection.
db = SQLAlchemy(application)


# Define database tables.
class IntegerSequencedItem(db.Model):
    __tablename__ = "integer_sequenced_items"
    id = db.Column(db.BigInteger().with_variant(db.Integer, "sqlite"), primary_key=True)
    sequence_id = db.Column(UUIDType(), nullable=False)
    position = db.Column(db.BigInteger(), nullable=False)
    topic = db.Column(db.String(255), nullable=False)
    state = db.Column(db.Text(), nullable=False)
    __table_args__ = (db.Index("index", "sequence_id", "position", unique=True),)


# Construct eventsourcing application.
@application.before_first_request
def init_example_application_with_sqlalchemy():
    init_example_application(
        entity_record_manager=SQLAlchemyRecordManager(
            record_class=IntegerSequencedItem, session=db.session
        )
    )


# Define Web application.
@application.route("/")
def hello():
    app = get_example_application()
    entity_id = app.create_new_example(foo="Hello There!").id
    entity = app.example_repository[entity_id]
    return "<h1 style='color:blue'>{}</h1>".format(entity.foo)


# Run directly, with uWSGI, or otherwise as a WSGI application.
#
# uwsgi -H PATH_TO_VIRTUALENV --master --processes 4 --threads 2 --wsgi-file PATH_TO_THIS_FILE --http :5001
#
if __name__ == "__main__":
    # Create tables.
    db.create_all()

    # Run the application.
    application.run(host="0.0.0.0", port=5001)