Database Programming in Python

By pjain      Published April 21, 2020, 7:43 p.m. in blog Programming   

SQLite Programming

Write to sqlite DB

$ chmod a+x my_rss_notifier.py $ sudo cp my_rss_notifier.py /etc/cron.hourly # https://en.wikipedia.org/wiki/Cron

1
#!/usr/bin/python3

import sqlite3 import feedparser

db = None feed_url = 'https://fedoramagazine.org/feed/'

def db_init(): db_connection = sqlite3.connect('/var/tmp/magazine_rss.sqlite') db = db_connection.cursor() db.execute('CREATE TABLE IF NOT EXISTS magazine (title TEXT, date TEXT)')

""" Check if a given pair of article title and date is in the database.
Args:
    article_title (str): The title of an article
    article_date  (str): The publication date of an article
Return:
    True if the article is not in the database
    False if the article is already present in the database
"""

def article_is_not_db(article_title, article_date): db.execute("SELECT * from magazine WHERE title=? AND date=?", (article_title, article_date)) if not db.fetchall(): return True else: return False

""" Add a new article title and date to the database
Args:
    article_title (str): The title of an article
    article_date (str): The publication date of an article
"""

def add_article_to_db(article_title, article_date): db.execute("INSERT INTO magazine VALUES (?,?)", (article_title, article_date)) db_connection.commit()

""" Get articles from RSS feed """

def read_article_feed(): feed = feedparser.parse(feed_url) for article in feed['entries']: if article_is_not_db(article['title'], article['published']): send_notification(article['title'], article['link']) add_article_to_db(article['title'], article['published'])

if name == 'main': db_init() read_article_feed() db_connection.close()

Write to DB text file with | delim

import time

print time.strftime("%a, %b %d %I:%M %p")

current_time_millis = lambda: int(round(time.time() * 1000)) current_timestamp = current_time_millis()

db = 'feeds.db' #'/var/www/radio/data/feeds.db' limit = 12 * 3600 * 1000 print("curtsms=",current_timestamp,"limit=",limit)

def post_is_in_db(title): with open(db, 'r') as database: for line in database: if title in line: return True return False

# return true if the title is in the database with a timestamp > limit

def post_is_in_db_with_old_timestamp(title): with open(db, 'r') as database: for line in database: if title in line: ts_as_string = line.split('|', 1)[1] ts = long(ts_as_string) if current_timestamp - ts > limit: return True return False

posts_to_print = [] posts_to_skip = []

for post in feed.entries: # if post is already in the database, skip it # TODO check the time title = post.title if post_is_in_db_with_old_timestamp(title): posts_to_skip.append(title) else: posts_to_print.append(title)

# add all the posts we're going to print to the database with the current timestamp
# (but only if they're not already in there)

f = open(db, 'a') for title in posts_to_print: if not post_is_in_db(title): f.write(title + "|" + str(current_timestamp) + "\n") print(title + "\n") f.close


0 comments

There are no comments yet

Add new comment

Similar posts

Sublime Text, Plugins and Custom Programming in Python

Debug Python

Python Applications and Frameworks 101

Python Scratch, Notes