78 lines
1.9 KiB
Python
78 lines
1.9 KiB
Python
import contextlib
|
|
from datetime import datetime
|
|
import logging
|
|
import os
|
|
import sqlite3
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
DATABASE_FILE = os.environ['DATABASE']
|
|
|
|
assert os.path.isfile(DATABASE_FILE)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def get_db():
|
|
db = sqlite3.connect(DATABASE_FILE)
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def get_name_and_replies(code):
|
|
with get_db() as db:
|
|
rows = db.execute(
|
|
'''\
|
|
WITH latest_reply AS (
|
|
SELECT
|
|
adults, children
|
|
FROM replies
|
|
WHERE family_code = :code
|
|
ORDER BY date DESC
|
|
LIMIT 1
|
|
)
|
|
SELECT
|
|
families.name,
|
|
(SELECT adults FROM latest_reply) AS adults,
|
|
(SELECT children FROM latest_reply) AS children
|
|
FROM families
|
|
WHERE families.code = :code
|
|
LIMIT 1;
|
|
''',
|
|
dict(code=code),
|
|
)
|
|
try:
|
|
return next(rows)
|
|
except StopIteration:
|
|
raise ValueError("Invalid code")
|
|
|
|
|
|
def record_reply(code, adults, children):
|
|
assert isinstance(adults, int)
|
|
assert isinstance(children, int)
|
|
assert 0 <= adults and 0 <= children
|
|
assert adults > 0 or children == 0
|
|
|
|
with get_db() as db:
|
|
# Insert update
|
|
cursor = db.cursor()
|
|
date = datetime.utcnow()
|
|
cursor.execute(
|
|
'''\
|
|
INSERT INTO replies(family_code, date, adults, children)
|
|
VALUES(:code, :date, :adults, :children);
|
|
''',
|
|
dict(code=code, date=date, adults=adults, children=children),
|
|
)
|
|
cursor.execute('COMMIT;')
|
|
logger.info(
|
|
"Recorded update for code=%r date=%r adults=%d children=%d",
|
|
code,
|
|
date,
|
|
adults,
|
|
children,
|
|
)
|