Updating other Tables Automatically in Flask SQLAlchemy-CodePudding

I am learning Flask/SQLAlchemy and am trying to create a simple database. Some of the tables can be computed based on the other tables, but I'm not quite sure where/how to add that logic.

As a simplified problem, let's say there are three tables in this database (* is primary key, ** is foreign key):

User: | id* | name | username | password |
Transactions: | id* | user_id** | date | quantity | value |
Holdings: | id* | user_id** | date | holdings |

I am then using flask-admin to manually add an entry in the Transactions table. Let's suppose that we added the entry: user_id = 1, date = '5/31/2022', quantity = 10, value = $30.

The goal here is that when this entry is added, Flask knows to automatically add another entry to the Holdings table by calculating some Python function. Let's suppose that Holdings.holdings = Transactions.quantity * Transactions.value, so that the goal is to automatically add the following entry: user_id = 1, date = '5/31/2022', holdings = $300 to the Holdings table.

Any suggestions on how to do this?

In Code:

from flask import Flask
from flask_login import UserMixin

app = Flask(__name__)
# Flask boilerplate code, register blueprints, etc

db = SQLAlchemy()

# Database Models

class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(128), nullable=False)
    username = db.Column(db.String(128), unique=True, nullable=False)
    password = db.Column(db.String(128), nullable=False)

class Transactions(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    user = db.relationship('User')

    date = db.Column(db.DateTime(timezone=True), nullable=False)
    transaction = db.Column(db.Float())

class Holdings(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    user = db.relationship('User')

    date = db.Column(db.DateTime(timezone=True), nullable=False)
    holdings = db.Column(db.Float())


db.init_app(app)

Update This is largely based on the logic of @jorzel answer, with some changes as I forgot to mention I am using flask_sqlalchemy.

@event.listens_for(Transactions, 'after_insert')
def UpdateHoldings(mapper, connection, transaction):

    # Adding new entry
    holdings_val = transaction.quantity * transaction.value
    holdings = Holdings(user_id = transaction.user_id,
                        date = transaction.date,
                        holdings = holdings_val)
    db.session.add( holdings )

    # (Not mentioned in OP, but if we want to update some table)
    # ( And assuming this table does not have date field )
    another_tbl = AnotherTable.__table__
    connection.execute(
        another_tbl.update().where(another_tbl.c.user_id == 
        transaction.user_id).values( holdings = holdings_val )
    )

Note: For some reason, db.session.commit throws a This session is closed error. However not including it seems to work.

CodePudding user response：

I have prepared a simple example using sqlalchemy event after_insert handling.

from datetime import datetime
from sqlalchemy import event, MetaData
from sqlalchemy.orm import sessionmaker
from tutorial.models import Transaction, Holding, User

DB_URI = "sqlite://"

engine = create_engine(DB_URI)
metadata = MetaData()
Base = declarative_base(metadata=metadata)


@contextmanager
def transaction_scope(session):
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise


def get_session(bind):
    Session = sessionmaker(bind)
    return Session()


@event.listens_for(Transaction, 'after_insert')
def receive_after_insert(mapper, connection, target):
    with get_session(connection) as session:
        with transaction_scope(session):
            transaction = target
            holding = Holding(
                user_id=transaction.user_id, 
                holdings=transaction.value * transaction.quantity, 
                date=transaction.date
            )
            session.add(holding)

def add_user(session) -> int:
    with transaction_scope(session):
        user = User(name='Test', username='test', password='secret')
        session.add(user)
        session.flush()
    return user.id

def tutorial_after_insert(with_raise=False):
    with get_session(engine) as session:
        user_id = add_user(session)
        with transaction_scope(session):
            transaction = Transaction(
                quantity=10,
                value=11.2,
                user_id=user_id,
                date=datetime.utcnow()
            )
            session.add(transaction)
            session.flush()
            if with_raise:
                raise Exception('exception')

try:
    tutorial_after_insert(with_raise=False)
except Exception as e:
    print(e)
finally:
    with get_session(engine) as session:
        print(session.query(Holding).all())
        print(session.query(Transaction).all())

You will get following result if no exception is raised:

[Holding(id=1, user_id=1, value=112.0, date=2022-06-01 08:27:26.307473)]
[Transaction(id=1, user_id=1, quantity=10, value=11.2, date=2022-06-01 08:27:26.307473)]

If an exception occurs, no record would be added.

Similar example I have added to this repo: https://github.com/jorzel/sqlalchemy-events-tutorial