in src/accounts/contacts/db.py [0:0]
def get_contacts(self, username):
"""Get a list of contacts for the specified username.
Params: username - the username of the user
Return: a list of contacts in the form of key/value attribute dicts,
[ {'label': contact1, ...}, {'label': contact2, ...}, ...]
Raises: SQLAlchemyError if there was an issue with the database
"""
contacts = list()
statement = self.contacts_table.select().where(
self.contacts_table.c.username == username
)
self.logger.debug("QUERY: %s", str(statement))
with self.engine.connect() as conn:
result = conn.execute(statement)
for row in result:
contact = {
"label": row["label"],
"account_num": row["account_num"],
"routing_num": row["routing_num"],
"is_external": row["is_external"],
}
contacts.append(contact)
self.logger.debug("RESULT: Fetched %d contacts.", len(contacts))
return contacts