Loading obakasan.py +31 −3 Changes for obakasan.py: 31 added lines, 3 removed lines. Original line number Diff line number Diff line #!/usr/bin/env python3 import pymysql import os from apscheduler.schedulers.blocking import BlockingScheduler from pymysql.cursors import DictCursor WP_POSTVIEWS_GET_COUNTS_SQL = """select p.ID, p.post_title, cast(m.meta_value as signed integer) as view_cnt, p.guid as url, p.post_type from wp_postmeta m, wp_posts p where m.meta_key = 'views' and m.post_id = p.ID order by view_cnt desc; """ def connect() -> pymysql.Connection: host, port_str = os.environ["DB_HOST"].split(":") connection = pymysql.connect( host=host, user=os.environ["DB_USER"], password=os.environ["DB_PASSWORD"], database=os.environ["DB_NAME"], port=int(port_str), cursorclass=DictCursor, ) def foo(): print("Hello World") def get_views() -> None: with connect() as connection: with connection.cursor() as cursor: cursor.execute(WP_POSTVIEWS_GET_COUNTS_SQL) print(cursor.fetchone()) if __name__ == "__main__": scheduler = BlockingScheduler() scheduler.add_job(foo, "interval", minutes=1) scheduler.add_job(get_views, "interval", minutes=1) scheduler.start() requirements.txt +1 −0 Changes for requirements.txt: 1 added line, 0 removed lines. Original line number Diff line number Diff line APScheduler~=3.10.4 PyMySQL~=1.1.1 Loading
obakasan.py +31 −3 Changes for obakasan.py: 31 added lines, 3 removed lines. Original line number Diff line number Diff line #!/usr/bin/env python3 import pymysql import os from apscheduler.schedulers.blocking import BlockingScheduler from pymysql.cursors import DictCursor WP_POSTVIEWS_GET_COUNTS_SQL = """select p.ID, p.post_title, cast(m.meta_value as signed integer) as view_cnt, p.guid as url, p.post_type from wp_postmeta m, wp_posts p where m.meta_key = 'views' and m.post_id = p.ID order by view_cnt desc; """ def connect() -> pymysql.Connection: host, port_str = os.environ["DB_HOST"].split(":") connection = pymysql.connect( host=host, user=os.environ["DB_USER"], password=os.environ["DB_PASSWORD"], database=os.environ["DB_NAME"], port=int(port_str), cursorclass=DictCursor, ) def foo(): print("Hello World") def get_views() -> None: with connect() as connection: with connection.cursor() as cursor: cursor.execute(WP_POSTVIEWS_GET_COUNTS_SQL) print(cursor.fetchone()) if __name__ == "__main__": scheduler = BlockingScheduler() scheduler.add_job(foo, "interval", minutes=1) scheduler.add_job(get_views, "interval", minutes=1) scheduler.start()
requirements.txt +1 −0 Changes for requirements.txt: 1 added line, 0 removed lines. Original line number Diff line number Diff line APScheduler~=3.10.4 PyMySQL~=1.1.1