Coverage for art_studio_tz/cli.py: 61%

1"""Command Line Interface (CLI) for quotes project.""" 

2import os 

3from io import StringIO 

4from sqlalchemy.engine import Engine 

5import pathlib 

6import rich 

7from rich.table import Table 

8from contextlib import contextmanager 

9from typing import List 

10 

11import art_studio_tz 

12 

13import typer 

14 

15 

16 

17app = typer.Typer(add_completion=True) 

18 

19 

20@app.command() 

21def version(): 

22 """Return version of app_quote application""" 

23 print(art_studio_tz.__version__) 

24 

25@app.command() 

26def start(url: str = typer.Option("https://zenquotes.io/api/random", "-u", "--url", help="URL for get quotes, default https://zenquotes.io/api/random"), 

27 pause: float = typer.Option(5, "-p", "--pause", help="pause between requests quotes in seconds, default 5 seconds")): 

28 """Get quote from url and add to db. with pause between requests. 

29 url - URL for get quotes, default https://zenquotes.io/api/random 

30 pause - pause between requests quotes in seconds, default 5 seconds""" 

31 with quote_db() as db: 

32 try: 

33 db.start(url, pause) 

34 except art_studio_tz.BadReqest: 

35 print(f"Could not get quote from {url}") 

36 

37@app.command("list") 

38def list_quote( 

39 author: str = typer.Option(None, "-a", "--author", help="sort for author") 

40): 

41 """ 

42 List quotes in db. 

43 """ 

44 with quote_db() as db: 

45 the_quote = db.list_quote(author=author) 

46 table = Table(box=rich.box.SIMPLE) 

47 table.add_column("ID") 

48 table.add_column("TimeStep") 

49 table.add_column("Quote") 

50 table.add_column("Author") 

51 for t in the_quote: 

52 author = "" if t.author is None else t.author 

53 table.add_row(str(t.id),t.timestep, t.text, author) 

54 out = StringIO() 

55 rich.print(table, file=out) 

56 print(out.getvalue()) 

57 

58@app.command() 

59def add( 

60 text: List[str], 

61 author: str = typer.Option(None, "-a", "--author", help="Author quote") 

62): 

63 """Add a quote to db.""" 

64 text = " ".join(text) if text else None 

65 with quote_db() as db: 

66 db.add_quote(art_studio_tz.Quote(text=text, author=author)) 

67 

68@app.command() 

69def delete(quote_id: int): 

70 """Remove quote in db with given id.""" 

71 with quote_db() as db: 

72 try: 

73 db.delete_quote(quote_id) 

74 except art_studio_tz.InvalidQuoteId: 

75 print(f"Error: Invalid qupte id {quote_id}") 

76 

77@app.command() 

78def update( 

79 quote_id: int, 

80 author: str = typer.Option(None, "-o", "--owner"), 

81 text: List[str] = typer.Option(None, "-t", "--text"), 

82): 

83 """Modify a quote in db with given id with new info.""" 

84 text = " ".join(text) if text else None 

85 with quote_db() as db: 

86 try: 

87 db.update_quote( 

88 quote_id, art_studio_tz.Quote(text=text, author=author) 

89 ) 

90 except art_studio_tz.InvalidQuoteId: 

91 print(f"Error: Invalid quote id {quote_id}") 

92 

93@app.command() 

94def config(): 

95 """List the path to the quotes db.""" 

96 with quote_db() as db: 

97 print(db.path()) 

98 

99 

100@app.command() 

101def count(): 

102 """Return number of quotes in db.""" 

103 with quote_db() as db: 

104 print(db.count()) 

105 

106@app.command() 

107def get(user: str = typer.Option(..., "-u", "--user", help="Database user"), 

108 password: str = typer.Option(..., "-p", "--password", help="Database password"), 

109 host: str = typer.Option("localhost", "-H", "--host", help="Database host, default localhost"), 

110 port: int = typer.Option(3306, "-P", "--port", help="Database port, default 3306"), 

111 database: str = typer.Option("quotes_db", "-d", "--database", help="Database name, default quotes_db"), 

112 url: str = typer.Option("https://zenquotes.io/api/quotes", "--url", help="URL for get quotes, default https://zenquotes.io/api/random"), 

113 ): 

114 """Get 50 quotes from url and add to mySQL""" 

115 with quote_db_sql(user=user, password=password, host=host, port=port, database=database) as db_sql: 

116 try: 

117 db_sql.get_some_quotes(url) 

118 except art_studio_tz.BadReqest: 

119 print(f"Could not get quote from {url}") 

120 

121@app.command() 

122def list_sql( 

123 user: str = typer.Option(..., "-u", "--user", help="Database user"), 

124 password: str = typer.Option(..., "-p", "--password", help="Database password"), 

125 host: str = typer.Option("localhost", "-H", "--host", help="Database host, default localhost"), 

126 port: int = typer.Option(3306, "-P", "--port", help="Database port, default 3306"), 

127 database: str = typer.Option("quotes_db", "-d", "--database", help="Database name, default quotes_db"), 

128 author: str = typer.Option(None, "-a", "--author", help="sort for author") 

129): 

130 """ 

131 List quotes in mySQL 

132 """ 

133 with quote_db_sql(user=user, password=password, host=host, port=port, database=database) as db_sql: 

134 the_quote = db_sql.list_quote(author=author) 

135 table = Table(box=rich.box.SIMPLE) 

136 table.add_column("ID") 

137 table.add_column("TimeStep") 

138 table.add_column("Quote") 

139 table.add_column("Author") 

140 for t in the_quote: 

141 author = "" if t.author is None else t.author 

142 table.add_row(str(t.id), str(t.timestep), t.text, author) 

143 out = StringIO() 

144 rich.print(table, file=out) 

145 print(out.getvalue()) 

146 

147@app.command() 

148def delete_all_sql(user: str = typer.Option(..., "-u", "--user", help="Database user"), 

149 password: str = typer.Option(..., "-p", "--password", help="Database password"), 

150 host: str = typer.Option("localhost", "-H", "--host", help="Database host, default localhost"), 

151 port: int = typer.Option(3306, "-P", "--port", help="Database port, default 3306"), 

152 database: str = typer.Option("quotes_db", "-d", "--database", help="Database name, default quotes_db"), 

153 ): 

154 """Delete all quotes in mySQL""" 

155 with quote_db_sql(user=user, password=password, host=host, port=port, database=database) as db_sql: 

156 try: 

157 db_sql.delete_all() 

158 except Exception as err: 

159 print(f"Could not delete quotes in {database} because {err}") 

160 

161@app.command() 

162def list_latest_5( 

163 user: str = typer.Option(..., "-u", "--user", help="Database user"), 

164 password: str = typer.Option(..., "-p", "--password", help="Database password"), 

165 host: str = typer.Option("localhost", "-H", "--host", help="Database host, default localhost"), 

166 port: int = typer.Option(3306, "-P", "--port", help="Database port, default 3306"), 

167 database: str = typer.Option("quotes_db", "-d", "--database", help="Database name, default quotes_db"), 

168 number: int = typer.Option(5, "-n", "--number", help="Number of latest quotes to list, default 5") 

169): 

170 """ 

171 list latest 'number' quotes in mySQL, default 5 

172 """ 

173 with quote_db_sql(user=user, password=password, host=host, port=port, database=database) as db_sql: 

174 the_quote = db_sql.get_latest(number) 

175 table = Table(box=rich.box.SIMPLE) 

176 table.add_column("ID") 

177 table.add_column("TimeStep") 

178 table.add_column("Quote") 

179 table.add_column("Author") 

180 for t in the_quote: 

181 author = "" if t.author is None else t.author 

182 table.add_row(str(t.id), str(t.timestep), t.text, author) 

183 out = StringIO() 

184 rich.print(table, file=out) 

185 print(out.getvalue()) 

186 

187@app.callback(invoke_without_command=True) 

188def main(ctx: typer.Context): 

189 """ 

190 quotes is a small command line task tracking application. 

191 """ 

192 if ctx.invoked_subcommand is None: 

193 list_quote(author=None) 

194 

195def get_path(): 

196 

197 db_path_env = os.getenv("QUOTES_DB_DIR", "") 

198 if db_path_env: 

199 db_path = pathlib.Path(db_path_env) 

200 else: 

201 db_path = pathlib.Path(os.getcwd()) 

202 return db_path 

203 

204 

205@contextmanager 

206def quote_db(): 

207 """Context manager for QuoteDB. 

208 

209 Yields: 

210 QuoteDB: QuoteDB instance 

211 """ 

212 db_path = get_path() 

213 db = art_studio_tz.QuoteDB(db_path) 

214 yield db 

215 

216@contextmanager 

217def quote_db_sql(user: str, password: str, host: str, port: int, database: str): 

218 """ 

219 Context manager for QuoteDBsql. 

220 

221 Args: 

222 user (str): username for database 

223 password (str): password for database 

224 host (str): hostname for database 

225 port (int): port for database 

226 database (str): database name 

227 

228 Yields: 

229 QuoteDBsql: QuoteDBsql instance 

230 """ 

231 db = art_studio_tz.QuoteDBsql(user=user, password=password, host=host, port=port, database=database) 

232 try: 

233 yield db 

234 finally: 

235 if hasattr(db, "engine") and isinstance(db.engine, Engine): 

236 db.engine.dispose()