-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmodels.py
67 lines (58 loc) · 2.02 KB
/
models.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
from sqlalchemy import create_engine, Column, Integer, String, DateTime, JSON
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
Base = declarative_base()
class Search(Base):
__tablename__ = 'searches'
id = Column(Integer, primary_key=True)
query = Column(String)
country = Column(String)
timestamp = Column(DateTime, default=datetime.now)
related_searches = Column(JSON)
settings = Column(JSON)
@classmethod
def create_tables(cls, engine):
Base.metadata.create_all(engine)
class SearchManager:
def __init__(self, db_path='sqlite:///searches.db'):
self.engine = create_engine(db_path)
Search.create_tables(self.engine)
self.Session = sessionmaker(bind=self.engine)
def save_search(self, query, country, related_searches, settings):
session = self.Session()
try:
search = Search(
query=query,
country=country,
related_searches=related_searches,
settings=settings
)
session.add(search)
session.commit()
return search
finally:
session.close()
def get_recent_searches(self, limit=10):
session = self.Session()
try:
return session.query(Search).order_by(Search.timestamp.desc()).limit(limit).all()
finally:
session.close()
def get_search_by_id(self, search_id):
session = self.Session()
try:
return session.query(Search).filter(Search.id == search_id).first()
finally:
session.close()
def delete_search(self, search_id):
session = self.Session()
try:
search = session.query(Search).filter(Search.id == search_id).first()
if search:
session.delete(search)
session.commit()
return True
return False
finally:
session.close()