rattail/rattail
2
0
Fork 0
Code Issues 7 Pull requests Actions Packages Projects Releases Wiki Activity

Add command logic functions for running reports, purging things

Browse source 
for use with typer commands
This commit is contained in:
Lance Edgar 2024-05-28 21:13:33 -05:00
parent a81a4b99d8
commit efbc97727b
2 changed files with 108 additions and 89 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

103
rattail/commands/purging.py
View file

@ -2,7 +2,7 @@
################################################################################ ################################################################################
# #
# Rattail -- Retail Software Framework # Rattail -- Retail Software Framework
# Copyright © 2010-2023 Lance Edgar # Copyright © 2010-2024 Lance Edgar
# #
# This file is part of Rattail. # This file is part of Rattail.
# #
@ -72,55 +72,15 @@ class PurgeSubcommand(Subcommand):
"at the end.") "at the end.")
def run(self, args): def run(self, args):
log.info("will purge things of type: %s", self.purge_title) run_purge(self.config, self.purge_title, self.purge_title_plural,
self.find_things_to_purge, self.purge_thing,
if args.before and args.before_days != self.default_before_days: before=args.before, before_days=args.before_days,
log.warning("specifying both --before and --before-days is " default_before_days=self.default_before_days,
"redundant; --before will take precedence.") dry_run=args.dry_run)
session = self.make_session()
# calculate our cutoff date
if args.before:
cutoff = args.before
else:
today = self.app.today()
cutoff = today - datetime.timedelta(days=args.before_days)
cutoff = datetime.datetime.combine(cutoff, datetime.time(0))
cutoff = self.app.localtime(cutoff)
log.info("using %s as cutoff date", cutoff.date())
# find things, and purge them
things = self.find_things_to_purge(session, cutoff,
dry_run=args.dry_run)
log.info("found %s thing(s) to purge", len(things or []))
if things:
purged = self.purge_things(session, things, cutoff,
dry_run=args.dry_run)
log.info("%spurged %s %s",
"(would have) " if args.dry_run else "",
purged, self.purge_title_plural)
self.finalize_session(session, dry_run=args.dry_run)
session.close()
def find_things_to_purge(self, session, cutoff, dry_run=False): def find_things_to_purge(self, session, cutoff, dry_run=False):
pass pass
def purge_things(self, session, things, cutoff, dry_run=False):
result = self.app.make_object(purged=0)
def purge(thing, i):
if self.purge_thing(session, thing, cutoff,
dry_run=dry_run):
result.purged += 1
if i % 200 == 0:
session.flush()
self.progress_loop(purge, things,
message="Purging {}".format(self.purge_title_plural))
return result.purged
def purge_thing(self, session, thing, cutoff, dry_run=False): def purge_thing(self, session, thing, cutoff, dry_run=False):
""" """
This method should contain logic which actually "purges" something. This method should contain logic which actually "purges" something.
@ -173,3 +133,54 @@ class PurgeExport(PurgeSubcommand):
shutil.rmtree(path) shutil.rmtree(path)
return True return True
def run_purge(config, purge_title, purge_title_plural, thing_finder, thing_purger,
before=None, before_days=None, default_before_days=90,
dry_run=False):
from rattail.db.util import finalize_session
log.info("will purge things of type: %s", purge_title)
if before and before_days:
log.warning("specifying both --before and --before-days is "
"redundant; --before will take precedence.")
app = config.get_app()
session = app.make_session()
# calculate our cutoff date
if before:
cutoff = before
else:
today = app.today()
cutoff = today - datetime.timedelta(days=before_days or default_before_days)
cutoff = datetime.datetime.combine(cutoff, datetime.time(0))
cutoff = app.localtime(cutoff)
log.info("using %s as cutoff date", cutoff.date())
# find things, and purge them
things = thing_finder(session, cutoff, dry_run=dry_run)
log.info("found %s thing(s) to purge", len(things or []))
if things:
purged = purge_things(config, session, things, thing_purger, cutoff, dry_run=dry_run)
log.info("%spurged %s %s",
"(would have) " if dry_run else "",
purged, purge_title_plural)
finalize_session(session, dry_run=dry_run)
def purge_things(config, session, things, purger, cutoff, dry_run=False):
app = config.get_app()
result = app.make_object(purged=0)
def purge(thing, i):
if purger(session, thing, cutoff, dry_run=dry_run):
result.purged += 1
if i % 200 == 0:
session.flush()
app.progress_loop(purge, things, progress,
message=f"Purging {purge_title_plural}")
return result.purged

94
rattail/commands/reports.py
View file

@ -2,7 +2,7 @@
################################################################################ ################################################################################
# #
# Rattail -- Retail Software Framework # Rattail -- Retail Software Framework
# Copyright © 2010-2023 Lance Edgar # Copyright © 2010-2024 Lance Edgar
# #
# This file is part of Rattail. # This file is part of Rattail.
# #
@ -56,54 +56,62 @@ class WeeklyReport(ReportSubcommand):
def run(self, args): def run(self, args):
self.ensure_report_key() self.ensure_report_key()
session = self.make_session() user = self.get_runas_user()
model = self.model run_weekly_report(self.config, self.report_key, self.simple_report_name,
email_key=self.email_key, user=user, progress=self.progress)
# first must determine most recent complete Mon-Sun date range
# TODO: should probably be more flexible about date range..
today = self.app.today()
sunday = get_sunday(today)
monday = get_monday(sunday)
start_date = monday
end_date = sunday
# determine naming for the report def run_weekly_report(config, report_key, simple_report_name, email_key=None, user=None,
report_name = "{} {} thru {}".format( progress=None):
self.simple_report_name, app = config.get_app()
start_date.strftime("%Y-%m-%d"), model = app.model
end_date.strftime("%Y-%m-%d")) session = app.make_session()
try: # first must determine most recent complete Mon-Sun date range
# see if this report has already been ran # TODO: should probably be more flexible about date range..
output = session.query(model.ReportOutput)\ today = app.today()
.filter(model.ReportOutput.report_type == self.report_key)\ sunday = get_sunday(today)
.filter(model.ReportOutput.report_name == report_name)\ monday = get_monday(sunday)
.one() start_date = monday
end_date = sunday
except orm.exc.NoResultFound: # determine naming for the report
report_name = "{} {} thru {}".format(
simple_report_name,
start_date.strftime("%Y-%m-%d"),
end_date.strftime("%Y-%m-%d"))
# generate report file and commit result try:
handler = self.app.get_report_handler() # see if this report has already been ran
report = handler.get_report(self.report_key) output = session.query(model.ReportOutput)\
params = {'start_date': start_date, 'end_date': end_date} .filter(model.ReportOutput.report_type == report_key)\
user = self.get_runas_user(session=session) .filter(model.ReportOutput.report_name == report_name)\
output = handler.generate_output(session, report, params, user, .one()
progress=self.progress)
session.commit()
# try to make url for report except orm.exc.NoResultFound:
report_url = None
base_url = self.config.base_url()
if base_url:
report_url = '{}/reports/generated/{}'.format(
base_url, output.uuid)
# send report output as email # generate report file and commit result
email_key = self.email_key or self.report_key handler = app.get_report_handler()
handler.email_output(report, output, email_key, report = handler.get_report(report_key)
extra_data={'report_url': report_url}) params = {'start_date': start_date, 'end_date': end_date}
if user:
user = session.get(model.User, user.uuid)
output = handler.generate_output(session, report, params, user,
progress=progress)
session.commit()
else: # try to make url for report
log.warning("report output already exists: %s", report_name) report_url = None
base_url = config.base_url()
if base_url:
report_url = '{}/reports/generated/{}'.format(
base_url, output.uuid)
session.close() # send report output as email
handler.email_output(report, output, email_key or report_key,
extra_data={'report_url': report_url})
else:
log.warning("report output already exists: %s", report_name)
session.close()