Miroslavsckaya/tg_rss_bot
1
1
Fork 0
Code Issues 7 Pull requests Projects 1 Releases Activity

add __convert_to_list_of_dicts method
Some checks failed
continuous-integration/drone/push Build is passing
Details
continuous-integration/drone/pr Build is failing
Details

Browse source
This commit is contained in:
mitsuha_s 2022-07-13 20:39:17 +00:00
parent 5d0a6843b6
commit feef909c61
1 changed files with 12 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
database.py
View file

@ -121,24 +121,24 @@ class Database:
subscribers = self.cur.fetchall() subscribers = self.cur.fetchall()
return list(map(lambda x: x['telegram_id'], subscribers)) return list(map(lambda x: x['telegram_id'], subscribers))
def find_feeds(self) -> list[DictRow]: def find_feeds(self) -> list[dict]:
"""Get a list of feeds.""" """Get a list of feeds."""
self.log.debug('find_feeds()') self.log.debug('find_feeds()')
self.cur.execute('SELECT * FROM feeds') self.cur.execute('SELECT * FROM feeds')
return self.cur.fetchall() return self.__convert_to_list_of_dicts(self.cur.fetchall())
def find_user_feeds(self, user_id: int) -> list[DictRow]: def find_user_feeds(self, user_id: int) -> list[dict]:
"""Return a list of feeds the user is subscribed to.""" """Return a list of feeds the user is subscribed to."""
self.log.debug('find_user_feeds(user_id=\'%s\')', user_id) self.log.debug('find_user_feeds(user_id=\'%s\')', user_id)
self.cur.execute('SELECT * FROM feeds WHERE id IN (SELECT feed_id FROM subscriptions WHERE user_id = %s)', self.cur.execute('SELECT * FROM feeds WHERE id IN (SELECT feed_id FROM subscriptions WHERE user_id = %s)',
[user_id]) [user_id])
return self.cur.fetchall() return self.__convert_to_list_of_dicts(self.cur.fetchall())
def find_feed_items(self, feed_id: int) -> list[DictRow]: def find_feed_items(self, feed_id: int) -> list[dict]:
"""Get last feed items.""" """Get last feed items."""
self.log.debug('find_feed_items(feed_id=\'%s\')', feed_id) self.log.debug('find_feed_items(feed_id=\'%s\')', feed_id)
self.cur.execute('SELECT * FROM feeds_last_items WHERE feed_id = %s', [feed_id]) self.cur.execute('SELECT * FROM feeds_last_items WHERE feed_id = %s', [feed_id])
return self.cur.fetchall() return self.__convert_to_list_of_dicts(self.cur.fetchall())
def find_feed_items_urls(self, feed_id: int) -> list[str]: def find_feed_items_urls(self, feed_id: int) -> list[str]:
"""Return urls last feed items""" """Return urls last feed items"""
@ -180,3 +180,8 @@ class Database:
' description TEXT' ' description TEXT'
')' ')'
) )
@staticmethod
def __convert_to_list_of_dicts(rows: list[DictRow]) -> list[dict]:
"""Convert list of DictRows to list of dicts"""
return list(map(lambda x: dict(x), rows))