Coverage for art_studio_tz/api.py: 81%

1""" 

2API for the quets project 

3""" 

4import requests 

5import time, os 

6from datetime import datetime, timezone 

7from dataclasses import asdict 

8from dataclasses import dataclass 

9from dataclasses import field 

10from pathlib import Path 

11 

12 

13from .db import DB 

14from .db_sql import DBsql 

15 

16 

17__all__ = [ 

18 "Quote", 

19 "QuoteDB", 

20 "QuoteDBsql", 

21 "QuoteException", 

22 "MissingText", 

23 "InvalidQuoteId", 

24 "BadReqest", 

25] 

26 

27@dataclass 

28class Quote: 

29 """model for a quote. 

30 """ 

31 text: str = None 

32 author: str = None 

33 timestep: str = None 

34 id: int = field(default=None, compare=False) 

35 

36 @classmethod 

37 def from_dict(cls, d): 

38 """Create a Quote from a dict. 

39 

40 Args: 

41 d (_type_): dict with quote data. 

42 

43 Returns: 

44 _type_: Quote instance 

45 """ 

46 return Quote(**d) 

47 

48 def to_dict(self): 

49 """Return a dict representation of the Quote.""" 

50 return asdict(self) 

51 

52 

53class QuoteException(Exception): 

54 pass 

55 

56 

57class MissingText(QuoteException): 

58 pass 

59 

60 

61class InvalidQuoteId(QuoteException): 

62 pass 

63 

64class BadReqest(QuoteException): 

65 pass 

66 

67 

68class QuoteDB: 

69 """A class to manage a database of quotes. API for quotes. 

70 """ 

71 def __init__(self, db_path): 

72 """initialize the QuoteDB instance. 

73 

74 Args: 

75 db_path (_type_): Path to the database file. 

76 """ 

77 self._db_path = db_path 

78 self._db = DB(db_path, "quotes", ["timestep",'text', 'author']) 

79 

80 def add_quote(self, quote: Quote) -> int: 

81 """Add a quote, return the id of quote. 

82 

83 Args: 

84 quote (Quote): Quote instance. 

85 

86 Raises: 

87 MissingText: if quote.text is None. 

88 

89 Returns: 

90 int: id of the added quote. 

91 """ 

92 if not quote.text: 

93 raise MissingText 

94 if quote.author is None: 

95 quote.author = "" 

96 id = self._db.create(quote.to_dict()) 

97 self._db.update(id, {"id": id, 

98 "timestep":datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S %Z') }) 

99 return id 

100 

101 def get_quote(self, quote_id: int) -> Quote: 

102 """Return a quote with given quote_id. 

103 

104 Args: 

105 quote_id (int): id of the quote. 

106 

107 Raises: 

108 InvalidQuoteId: if quote_id is not in db. 

109 

110 Returns: 

111 Quote: Quote instance. 

112 """ 

113 db_item = self._db.read(quote_id) 

114 if db_item is not None: 

115 return Quote.from_dict(db_item) 

116 else: 

117 raise InvalidQuoteId(quote_id) 

118 

119 def list_quote(self, author=None): 

120 """Return a list of quotes. 

121 

122 Args: 

123 author (_type_, optional): Author name to filter quotes. Defaults to None. 

124 

125 Returns: 

126 List(Quote)): List of Quote instances. 

127 """ 

128 all = self._db.read_all() 

129 if author is not None: 

130 return [ 

131 Quote.from_dict(t) 

132 for t in all 

133 if (t["author"] == author) 

134 ] 

135 else: 

136 return [Quote.from_dict(t) for t in all] 

137 

138 def count(self) -> int: 

139 """Return the number of quotes in db.""" 

140 return self._db.count() 

141 

142 def update_quote(self, quote_id: int, quote_mods: Quote) -> None: 

143 """update a quote with modifications. 

144 

145 Args: 

146 quote_id (int): id of the quote to update. 

147 quote_mods (Quote): Quote instance with modifications. 

148 

149 Raises: 

150 InvalidQuoteId: if quote_id is not in db. 

151 """ 

152 try: 

153 self._db.update(quote_id, quote_mods.to_dict()) 

154 except KeyError as exc: 

155 raise InvalidQuoteId(quote_id) from exc 

156 

157 def start(self, url: str, pause: float): 

158 """get quotes from url and add them to db every 'pause' seconds. 

159 

160 Args: 

161 url (str): _url to get quotes from if not given use "https://zenquotes.io/api/random" 

162 pause (float): pause between requests in seconds. 

163 """ 

164 

165 

166 # url = "https://zenquotes.io/api/random"  

167 

168 print("For stop taking quotes press 'Ctrl + C'") 

169 

170 while True: 

171 try: 

172 response = requests.get(url) 

173 response.raise_for_status() # Генерирует исключение для статуса ошибки HTTP  

174 data = response.json() 

175 

176 if data: 

177 quote = data[0]["q"] 

178 author = data[0]["a"] 

179 self.add_quote(Quote(text=quote, author=author)) 

180 print(f"Added Quote: {quote} - Author: {author}") 

181 else: 

182 print("No data received.") 

183 print("For stop taking quotes press 'Ctrl + C'") 

184 time.sleep(pause) # Пауза в 1 секунду 

185 

186 except requests.exceptions.RequestException as e: 

187 print(f"Error fetching quote: {e}") 

188 

189 except KeyboardInterrupt: 

190 print("\nОстановка запроса цитат пользователем.") 

191 break 

192 

193 def delete_quote(self, quote_id: int) -> None: 

194 """Delete a quote with given quote_id. 

195 

196 Args: 

197 quote_id (int): id of the quote to delete. 

198 

199 Raises: 

200 InvalidQuoteId: if quote_id is not in db. 

201 """ 

202 try: 

203 self._db.delete(quote_id) 

204 except KeyError as exc: 

205 raise InvalidQuoteId(quote_id) from exc 

206 

207 def delete_all(self) -> None: 

208 """Remove all quotes from db.""" 

209 self._db.delete_all() 

210 

211 def path(self): 

212 """Return the path to the database file.""" 

213 return self._db_path 

214 

215class QuoteDBsql: 

216 """A class to manage a database of quotes using MySQL. API for quotes. 

217 """ 

218 def __init__(self, user, password, host = "localhost", port = 3306, database = "quotes_db"): 

219 """initialize the QuoteDBsql instance. 

220 

221 Args: 

222 user (str): username for the database. 

223 password (str): password for the database. 

224 host (str, optional): hpstname for the database. Defaults to "localhost". 

225 port (int, optional): port for the database. Defaults to 3306. 

226 database (str, optional): database name. Defaults to "quotes_db". 

227 """ 

228 self._db = DBsql(user, password, host, port, database) 

229 

230 def add_quote_sql(self, quote: Quote) -> None: 

231 """Add a quote to the database. 

232 

233 Args: 

234 quote (Quote): Quote instance. 

235 

236 Raises: 

237 MissingText: if quote.text is None. 

238 """ 

239 if not quote.text: 

240 raise MissingText 

241 if quote.author is None: 

242 quote.author = "" 

243 self._db.create(quote.to_dict()) 

244 

245 

246 def list_quote(self, author=None) -> list[Quote]: 

247 """Return a list of quotes. 

248 

249 Args: 

250 author (str, optional): Author name to filter quotes. Defaults to None. 

251 

252 Returns: 

253 list[Quote]: List of Quote instances. 

254 """ 

255 all = self._db.read_all() 

256 if author is not None: 

257 return [ 

258 Quote.from_dict(t) 

259 for t in all 

260 if (t["author"] == author) 

261 ] 

262 else: 

263 return [Quote.from_dict(t) for t in all] 

264 

265 

266 def get_some_quotes(self, url: str): 

267 """Get 50 quotes from url and add them to db. 

268 

269 Args: 

270 url (str): _url to get quotes from if not given use "https://zenquotes.io/api/quotes" 

271 """ 

272 

273 # url = "https://zenquotes.io/api/quotes"  

274 

275 try: 

276 response = requests.get(url) 

277 response.raise_for_status() 

278 data = response.json() 

279 if data: 

280 for item in data: 

281 quote = item["q"] 

282 author = item["a"] 

283 self.add_quote_sql(Quote(text=quote, author=author)) 

284 print(f"Added Quote: {quote} - Author: {author}") 

285 time.sleep(0.5) 

286 print('Added 50 quotes, you can see them using the "art_studio_t list_sql" command') 

287 else: 

288 print("No data received.") 

289 except requests.exceptions.RequestException as err: 

290 print(f"Error fetching quote: {err}") 

291 

292 def get_latest(self, number) -> list[Quote]: 

293 """Return the latest 'number' quotes default 5 

294 

295 Args: 

296 number (_type_): number of quotes to return. 

297 

298 Returns: 

299 list[Quote]: List of Quote instances. 

300 """ 

301 all = self._db.get_latest(number) 

302 return [Quote.from_dict(t) for t in all] 

303 

304 def delete_all(self) -> None: 

305 """Remove all quotes from db mySQL. 

306 """ 

307 self._db.delete_all() 

308 

309# todo: uncomment for testing 

310# if __name__ == "__main__": 

311# ob = QuoteDB(Path(os.getcwd())) 

312# # ob.start("https://zenquotes.io/api/random",10) 

313# print(ob.list_quote()) 

314# print(ob.count()) 

315# # ob.delete_all() 

316 

317# ob.delete_quote(1) 

318# ob.update_qote(2, Quote(text="New text")) 

319# print(ob.get_quote(2)) 

320# ob.add_quote(Quote(text="Some text", author="Some author")) 

321# print(ob.list_quote())  

322# print(ob.count()) 

323# ob.delete_all()  

324