Movie watchlist database.py diff between SQLite and PostgreSQL
建立於 差異永不過期
import os
import datetime
import datetime
import sqlite3
import psycopg2
from dotenv import load_dotenv
load_dotenv()
CREATE_MOVIES_TABLE = """CREATE TABLE IF NOT EXISTS movies (
CREATE_MOVIES_TABLE = """CREATE TABLE IF NOT EXISTS movies (
id INTEGER PRIMARY KEY,
id SERIAL PRIMARY KEY,
name TEXT,
name TEXT,
release_timestamp REAL
release_timestamp REAL
);"""
);"""
CREATE_USERS_TABLE = """CREATE TABLE IF NOT EXISTS users (
CREATE_USERS_TABLE = """CREATE TABLE IF NOT EXISTS users (
username TEXT PRIMARY KEY
username TEXT PRIMARY KEY
);"""
);"""
CREATE_WATCHED_TABLE = """CREATE TABLE IF NOT EXISTS watched (
CREATE_WATCHED_TABLE = """CREATE TABLE IF NOT EXISTS watched (
user_username TEXT,
user_username TEXT,
movie_id INTEGER,
movie_id INTEGER,
FOREIGN KEY(user_username) REFERENCES users(username),
FOREIGN KEY(user_username) REFERENCES users(username),
FOREIGN KEY(movie_id) REFERENCES movies(id)
FOREIGN KEY(movie_id) REFERENCES movies(id)
);"""
);"""
INSERT_MOVIE = "INSERT INTO movies (name, release_timestamp) VALUES (?, ?)"
INSERT_MOVIE = "INSERT INTO movies (name, release_timestamp) VALUES (%s, %s)"
SELECT_ALL_MOVIES = "SELECT * FROM movies;"
SELECT_ALL_MOVIES = "SELECT * FROM movies;"
SELECT_UPCOMING_MOVIES = "SELECT * FROM movies WHERE release_timestamp > ?;"
SELECT_UPCOMING_MOVIES = "SELECT * FROM movies WHERE release_timestamp > %s;"
INSERT_USER = "INSERT INTO users (username) VALUES (?)"
INSERT_USER = "INSERT INTO users (username) VALUES (%s)"
INSERT_WATCHED_MOVIE = "INSERT INTO watched (user_username, movie_id) VALUES (?, ?)"
INSERT_WATCHED_MOVIE = "INSERT INTO watched (user_username, movie_id) VALUES (%s, %s)"
SELECT_WATCHED_MOVIES = """SELECT movies.*
SELECT_WATCHED_MOVIES = """SELECT movies.*
FROM users
FROM users
JOIN watched ON users.username = watched.user_username
JOIN watched ON users.username = watched.user_username
JOIN movies ON watched.movie_id = movies.id
JOIN movies ON watched.movie_id = movies.id
WHERE users.username = ?;"""
WHERE users.username = %s;"""
SEARCH_MOVIE = """SELECT * FROM movies WHERE name LIKE ?;"""
SEARCH_MOVIE = """SELECT * FROM movies WHERE name LIKE %s;"""
connection = sqlite3.connect("data.db")
# Remember to not store the database URI in your code!
connection = psycopg2.connect(os.environ.get("DATABASE_URI"))
def create_tables():
def create_tables():
with connection:
with connection:
connection.execute(CREATE_MOVIES_TABLE)
with connection.cursor() as cursor:
connection.execute(CREATE_USERS_TABLE)
cursor.execute(CREATE_MOVIES_TABLE)
connection.execute(CREATE_WATCHED_TABLE)
cursor.execute(CREATE_USERS_TABLE)
cursor.execute(CREATE_WATCHED_TABLE)
def add_movie(name, release_timestamp):
def add_movie(name, release_timestamp):
with connection:
with connection:
connection.execute(INSERT_MOVIE, (name, release_timestamp))
with connection.cursor() as cursor:
cursor.execute(INSERT_MOVIE, (name, release_timestamp))
def get_movies(upcoming=False):
def get_movies(upcoming=False):
with connection:
with connection:
cursor = connection.cursor()
with connection.cursor() as cursor:
if upcoming:
if upcoming:
today_timestamp = datetime.datetime.today().timestamp()
today_timestamp = datetime.datetime.today().timestamp()
cursor.execute(SELECT_UPCOMING_MOVIES, (today_timestamp,))
cursor.execute(SELECT_UPCOMING_MOVIES, (today_timestamp,))
else:
else:
cursor.execute(SELECT_ALL_MOVIES)
cursor.execute(SELECT_ALL_MOVIES)
return cursor.fetchall()
return cursor.fetchall()
def add_user(username):
def add_user(username):
with connection:
with connection:
connection.execute(INSERT_USER, (username,))
with connection.cursor() as cursor:
cursor.execute(INSERT_USER, (username,))
def watch_movie(username, movie_id):
def watch_movie(username, movie_id):
with connection:
with connection:
connection.execute(INSERT_WATCHED_MOVIE, (username, movie_id))
with connection.cursor() as cursor:
cursor.execute(INSERT_WATCHED_MOVIE, (username, movie_id))
def get_watched_movies(username):
def get_watched_movies(username):
with connection:
with connection:
cursor = connection.cursor()
with connection.cursor() as cursor:
cursor.execute(SELECT_WATCHED_MOVIES, (username,))
cursor.execute(SELECT_WATCHED_MOVIES, (username,))
return cursor.fetchall()
return cursor.fetchall()
def search_movies(search_term):
def search_movies(search_term):
with connection:
with connection:
cursor = connection.cursor()
with connection.cursor() as cursor:
cursor.execute(SEARCH_MOVIE, (f"%{search_term}%",))
cursor.execute(SEARCH_MOVIE, (f"%{search_term}%",))
return cursor.fetchall()
return cursor.fetchall()