100 lines
2.6 KiB
Python
100 lines
2.6 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_people_and_replies(code):
|
|
with get_db() as db:
|
|
return list(db.execute(
|
|
'''\
|
|
SELECT
|
|
persons.id, persons.name,
|
|
(
|
|
SELECT rsvp FROM replies
|
|
WHERE
|
|
replies.person_id = persons.id
|
|
AND replies.date = (
|
|
SELECT MAX(date)
|
|
FROM replies
|
|
WHERE person_id = persons.id
|
|
)
|
|
) AS rsvp
|
|
FROM
|
|
persons
|
|
WHERE
|
|
persons.code = :code;
|
|
''',
|
|
dict(code=code),
|
|
))
|
|
|
|
|
|
def record_reply(code, persons):
|
|
assert isinstance(persons, dict)
|
|
with get_db() as db:
|
|
# Check the IDs of people match
|
|
expected_people_ids = set(
|
|
row[0]
|
|
for row in db.execute(
|
|
'''\
|
|
SELECT
|
|
id
|
|
FROM
|
|
persons
|
|
WHERE
|
|
code = :code;
|
|
''',
|
|
dict(code=code),
|
|
)
|
|
)
|
|
if not expected_people_ids:
|
|
logger.warning("Got reply with invalid code %r", code)
|
|
raise ValueError("Invalid code")
|
|
for person_id in persons.keys():
|
|
if person_id not in expected_people_ids:
|
|
logger.warning(
|
|
"Got invalid person id=%d for code %r",
|
|
person_id,
|
|
code,
|
|
)
|
|
raise ValueError("Invalid person")
|
|
|
|
# Insert update
|
|
cursor = db.cursor()
|
|
date = datetime.utcnow()
|
|
for person_id, rsvp in persons.items():
|
|
if rsvp not in ('yes', 'no'):
|
|
raise ValueError("Invalid rsvp %r", rsvp)
|
|
cursor.execute(
|
|
'''\
|
|
INSERT INTO replies(person_id, date, rsvp)
|
|
VALUES(:id, :date, :rsvp);
|
|
''',
|
|
dict(id=person_id, date=date, rsvp=rsvp),
|
|
)
|
|
cursor.execute('COMMIT;')
|
|
logger.info(
|
|
"Recorded update for code=%r date=%r persons: %s",
|
|
code,
|
|
date,
|
|
' '.join('%d=%s' % p for p in sorted(persons.items())),
|
|
)
|