In order to make this very dynamic, I have to switch to python celery. Using repeat_every will work alright but assuming an upgrade is done to your server, you need to be able to have control over the job running period of time. With celery, you can control the time your job runs. So following the folder module structure from celery docs, I have the following module:
app/celeryJobs/
__init__py
tasks.py
celeryWorker.py
billing.py
and having the following lines of code.
tasks.py
from app.celeryJobs.celeryWorker import celery, celery_log
from app.emails import billingEmail
import time
import json
from json import JSONEncoder
from sqlalchemy.orm import Session
from app.emails import billingEmail
from app.celeryJobs import billing
from app.utils import database
@celery.task(name ='app.celeryJobs.tasks.bill_creation')
def bill_creation():
billing.generateBills()
return {"message": "Bill Generation runs"}
@celery.task(name ='app.celeryJobs.tasks.sum_number')
def sum_number(x, y):
add = x + y
celery_log.info("I am suppose to run")
return add
@celery.task(name ='app.celeryJobs.tasks.sendBillingEmail')
def sendBillingEmail(email:str, fullName: str, month: str, amount:str):
time.sleep(5)
celery_log.info("I am suppose to run")
respon = billingEmail.sendBilling( email, fullName, month, amount)
json_object = json.dumps(dict(respon), indent = 4)
return json_object
celeryWorker.py
from celery import Celery
from celery.utils.log import get_task_logger
from celery.schedules import crontab
celery = Celery('app.celeryJobs',
broker="pyamqp://guest@localhost//",
include=['app.celeryJobs.tasks']
)
celery.conf.update(
task_serializer='json',
result_serializer='json',
timezone='UTC',
enable_utc=True,
imprts=("app.celeryJobs")
)
celery_log = get_task_logger(__name__)
celery.conf.beat_schedule = {
'sum_number': {
'task': 'app.celeryJobs.tasks.sum_number',
'schedule': 30.0,
'args': (16, 16)
},
'bill_creation': {
'task': 'app.celeryJobs.tasks.bill_creation',
'schedule': crontab(minute=0,hour='13,14,15,16,17,18,19,20, 21, 22, 23, 0'),
},
}
billing.py
from sqlalchemy.orm import Session
from app.models import model, schemas
from datetime import datetime, timedelta
from sqlalchemy import and_
from app.emails import billingEmail
import calendar
from app.celeryJobs import tasks
from sqlalchemy import create_engine
from sqlalchemy import Column, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.utils import config
import logging
def generateBills():
#Database Connection
db_string = config.settings.SQLALCHEMY_DATABASE_URI
con = create_engine(db_string)
Session = sessionmaker(con)
db = Session()
fifteen_days_ago = datetime.today() - timedelta(days = 15)
sensors = db.query( model.Sensor).all()
print('days',fifteen_days_ago)
print('leng sensors', len(sensors))
volumeUsed = 0
rate= db.query(model.Rate).order_by(model.Rate.id.desc()).first()
print('rate', rate.ratePerLitre)
for sensor in sensors:
senData = sensor
readings = db.query(model.SensorReadings).filter(model.SensorReadings.sensorId == senData.id and model.SensorReadings.dateAdded <= fifteen_days_ago).all()
user = db.query(model.User).filter(model.User.sensorId == senData.id).first()
print('leng readings',len(readings))
volumeUsed = sum(reading.volume for reading in readings)
print('volume',volumeUsed)
totalCost = volumeUsed * rate.ratePerLitre
print('id', senData.id)
print('totalcost', totalCost)
new_cost = model.Cost(sensorId = senData.id,
volumeUsed = volumeUsed,
ratePerLitre = rate.ratePerLitre,
totalCost = totalCost)
print(new_cost)
# db.add(new_cost)
# db.commit()
# db.refresh(new_cost)
if user is not None:
datem = datetime.strptime(str(senData.dateAdded), "%Y-%m-%d %H:%M:%S.%f")
month = calendar.month_name[datem.month]
print('month', month)
print('username',user.email)
# respon = billingEmail.sendBilling( user.email, user.fullName, month, str(totalCost))
# print(respon)
task = tasks.sendBillingEmail.delay( user.email, user.fullName, month, str(totalCost))
db.close()
logging.info("Cost bill generate for sensor")
Make sure your python virtual environment is activated, run the following commands on different terminals to run the celery periodic task.
celery -A app.celeryJobs.celeryWorker beat --loglevel=info
celery -A app.celeryJobs.celeryWorker worker -l info -P eventlet
Ensure you read the celery docs for details on setting up Broker and Backend.
Enjoy!
Alternatively, to use repeat_every and avoid the Depends' object has no attribute 'query' error.
Make sure you connect to the database directly as show in the lines of code below:
billing.py
from sqlalchemy.orm import Session
from app.models import model, schemas
from datetime import datetime, timedelta
from sqlalchemy import and_
from app.emails import billingEmail
import calendar
from app.celeryJobs import tasks
from sqlalchemy import create_engine
from sqlalchemy import Column, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.utils import config
import logging
def generateBills():
db_string = config.settings.SQLALCHEMY_DATABASE_URI
con = create_engine(db_string)
Session = sessionmaker(con)
db = Session()
fifteen_days_ago = datetime.today() - timedelta(days = 15)
sensors = db.query( model.Sensor).all()
print('days',fifteen_days_ago)
print('leng sensors', len(sensors))
volumeUsed = 0
rate= db.query(model.Rate).order_by(model.Rate.id.desc()).first()
print('rate', rate.ratePerLitre)
for sensor in sensors:
senData = sensor
readings = db.query(model.SensorReadings).filter(model.SensorReadings.sensorId == senData.id and model.SensorReadings.dateAdded <= fifteen_days_ago).all()
user = db.query(model.User).filter(model.User.sensorId == senData.id).first()
print('leng readings',len(readings))
volumeUsed = sum(reading.volume for reading in readings)
print('volume',volumeUsed)
totalCost = volumeUsed * rate.ratePerLitre
print('id', senData.id)
print('totalcost', totalCost)
new_cost = model.Cost(sensorId = senData.id,
volumeUsed = volumeUsed,
ratePerLitre = rate.ratePerLitre,
totalCost = totalCost)
print(new_cost)
# db.add(new_cost)
# db.commit()
# db.refresh(new_cost)
if user is not None:
datem = datetime.strptime(str(senData.dateAdded), "%Y-%m-%d %H:%M:%S.%f")
month = calendar.month_name[datem.month]
print('month', month)
print('username',user.email)
# respon = billingEmail.sendBilling( user.email, user.fullName, month, str(totalCost))
# print(respon)
task = tasks.sendBillingEmail.delay( user.email, user.fullName, month, str(totalCost))
db.close()
logging.info("Cost bill generate for sensor")
Then in your main.py
use the repeat_every to call the function as shown below:
@app.on_event("startup")
@repeat_every(seconds= 60*60*7, wait_first= True)
def billCreation():
print("Billing generation is running every 7 days")
billing.generateBills()
