-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathget_database.py
83 lines (69 loc) · 2.31 KB
/
get_database.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
# database.py
from sqlalchemy import create_engine, Column, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import func
import random
DATABASE_URL = 'sqlite:///quotes.db'
engine = create_engine(DATABASE_URL)
Session = sessionmaker(bind=engine)
Base = declarative_base()
class Quote(Base):
__tablename__ = 'quotes'
uid = Column(String, primary_key=True)
quote = Column(String)
author = Column(String)
Base.metadata.create_all(engine)
_session_factory = sessionmaker(bind=engine)
def _get_session():
"""Return a new Session object for each call"""
return _session_factory()
def add_quote(uid, quote, author):
"""Add a quote if the UID is not in the database yet"""
session = _get_session()
if not session.query(Quote).filter_by(uid=uid).first():
new_quote = Quote(uid=uid, quote=quote, author=author)
session.add(new_quote)
session.commit()
return True
return False
def delete_quote(uid):
"""Delete a quote by its UID"""
session = _get_session()
quote_obj = session.query(Quote).filter_by(uid=uid).first()
if quote_obj:
session.delete(quote_obj)
session.commit()
return True
return False
def fetch_quote_by_uid(uid):
"""Fetch a quote by its UID"""
session = _get_session()
return session.query(Quote).filter_by(uid=uid).first()
def fetch_random_quote():
"""Fetch a random quote"""
session = _get_session()
count = session.query(Quote).count()
random_index = random.randint(1, count)
return session.query(Quote).offset(random_index-1).limit(1).first()
def edit_quote(uid, quote=None, author=None):
"""Edit a quote by its UID"""
session = _get_session()
quote_obj = session.query(Quote).filter_by(uid=uid).first()
if quote_obj:
if quote:
quote_obj.quote = quote
if author:
quote_obj.author = author
session.commit()
return True
return False
def fetch_multiple_quotes(count=10):
"""Fetch multiple quotes"""
session = _get_session()
return session.query(Quote).order_by(func.random()).limit(count).all()
def get_quote_count():
"""Get the total count of quotes"""
session = _get_session()
query = session.query(Quote)
return query.count()