122 lines
3.9 KiB
Python
122 lines
3.9 KiB
Python
"""Module containing scheduled operation related models."""
|
|
|
|
# vim: set tw=80 ts=4 sw=4 sts=4:
|
|
|
|
from calendar import monthrange
|
|
|
|
import arrow
|
|
|
|
# pylint: disable=no-member,too-few-public-methods,too-many-instance-attributes
|
|
from . import db
|
|
|
|
from .accounts import Account
|
|
from .operations import Operation
|
|
|
|
|
|
class ScheduledOperation(db.Model):
|
|
"""Class used to handle scheduled operations."""
|
|
# pylint: disable=invalid-name
|
|
id = db.Column(db.Integer, primary_key=True)
|
|
|
|
start_date = db.Column(db.Date, nullable=False)
|
|
stop_date = db.Column(db.Date, nullable=False)
|
|
|
|
day = db.Column(db.Integer, nullable=False)
|
|
frequency = db.Column(db.Integer, nullable=False)
|
|
|
|
label = db.Column(db.String(500), nullable=False)
|
|
value = db.Column(db.Numeric(15, 2), nullable=False)
|
|
|
|
category = db.Column(db.String(100), nullable=True)
|
|
|
|
account_id = db.Column(
|
|
db.Integer,
|
|
db.ForeignKey('account.id')
|
|
)
|
|
|
|
account = db.relationship(
|
|
Account,
|
|
backref=db.backref('scheduled_operation', lazy="dynamic")
|
|
)
|
|
|
|
# pylint: disable=too-many-arguments
|
|
def __init__(self, start_date, stop_date, day, frequency, label, value,
|
|
account_id, category=None):
|
|
self.start_date = start_date
|
|
self.stop_date = stop_date
|
|
self.day = day
|
|
self.frequency = frequency
|
|
self.label = label
|
|
self.value = value
|
|
self.account_id = account_id
|
|
self.category = category
|
|
|
|
@classmethod
|
|
def query(cls):
|
|
"""Return base query for this class."""
|
|
return db.session.query(
|
|
cls
|
|
).order_by(
|
|
db.desc(cls.day),
|
|
cls.value,
|
|
cls.label,
|
|
)
|
|
|
|
def reschedule(self):
|
|
"""Reschedule a scheduled operation."""
|
|
# 1) delete unconfirmed operations for this account.
|
|
db.session.query(
|
|
Operation
|
|
).filter(
|
|
Operation.scheduled_operation_id == self.id,
|
|
Operation.confirmed == db.false()
|
|
).delete()
|
|
|
|
# 2) schedule remaining operations.
|
|
# Find the first date to have all dates in the range with the right
|
|
# day.
|
|
start_date = arrow.get(self.start_date)
|
|
day = min(self.day, monthrange(start_date.year, start_date.month)[1])
|
|
|
|
# First date is in the next month is below the start date day.
|
|
if day < start_date.day:
|
|
start_date = start_date.replace(months=+1)
|
|
start_date = start_date.replace(day=day)
|
|
|
|
# Stop date remains the same.
|
|
stop_date = arrow.get(self.stop_date)
|
|
|
|
# Iterate over the range.
|
|
date_range = arrow.Arrow.range("month", start_date, stop_date)
|
|
dates = date_range[::self.frequency]
|
|
for date in dates:
|
|
# If a date day is below the first date day, next dates will not
|
|
# have the right day.
|
|
day = min(self.day, monthrange(
|
|
date.year, date.month)[1])
|
|
|
|
date = date.replace(day=day)
|
|
|
|
# Search if a close operation exists.
|
|
|
|
if not db.session.query(Operation).filter(
|
|
Operation.account_id == self.account_id,
|
|
Operation.scheduled_operation_id == self.id,
|
|
Operation.operation_date >= date.replace(weeks=-2).date(),
|
|
Operation.operation_date <= date.replace(weeks=+2).date()
|
|
).count():
|
|
# Create not confirmed operation if not found.
|
|
operation = Operation(
|
|
operation_date=date.date(),
|
|
label=self.label,
|
|
value=self.value,
|
|
category=self.category,
|
|
account_id=self.account_id,
|
|
scheduled_operation_id=self.id,
|
|
pointed=False,
|
|
confirmed=False,
|
|
canceled=False
|
|
)
|
|
|
|
db.session.add(operation)
|