Coverage for src/database.mys : 99%
![Show keyboard shortcuts](keybd_closed.png)
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from sqlite import Database as SqliteDatabase
2from sqlite import Statement
3from os.path import Path
4from fiber import Lock
6class Package:
7 package_id: i64
8 name: string
9 latest_release: Release?
10 releases: [Release]
11 token: string
12 number_of_downloads: i64
13 builds: string
15class Release:
16 release_id: i64
17 version: string
18 description: string
20class Mys:
21 latest_release: MysRelease?
22 token: string
24class MysRelease:
25 release_id: i64
26 version: string
28class Database:
29 _database: SqliteDatabase
30 _create_mys: Statement
31 _modify_mys: Statement
32 _get_mys: Statement
33 _get_mys_release: Statement
34 _get_mys_release_by_id: Statement
35 _add_mys_release: Statement
36 _create_package: Statement
37 _delete_package: Statement
38 _delete_all_package_releases: Statement
39 _modify_package: Statement
40 _increment_package_download_count: Statement
41 _set_package_builds: Statement
42 _get_package: Statement
43 _get_packages: Statement
44 _get_package_release: Statement
45 _get_package_release_by_id: Statement
46 _get_package_releases: Statement
47 _add_package_release: Statement
48 _get_dependents: Statement
49 _remove_dependents: Statement
50 _add_dependent: Statement
51 _clear_activities: Statement
52 _add_activity: Statement
53 _get_activities: Statement
54 root_directory: Path
55 _lock: Lock
57 func __init__(self, root_directory: Path):
58 self.root_directory = root_directory
59 self._lock = Lock()
60 root_directory.mkdir(exists_ok=True)
62 self._database = SqliteDatabase(self.make_path("website.sqlite"))
64 self._database.execute("CREATE TABLE IF NOT EXISTS mys("
65 "mys_id INTEGER PRIMARY KEY,"
66 "latest_release_id INTEGER DEFAULT -1,"
67 "token TEXT NOT NULL"
68 ")")
69 self._database.execute("CREATE TABLE IF NOT EXISTS mys_releases("
70 "release_id INTEGER PRIMARY KEY,"
71 "version TEXT NOT NULL UNIQUE"
72 ")")
73 self._database.execute("CREATE TABLE IF NOT EXISTS packages("
74 "package_id INTEGER PRIMARY KEY,"
75 "name TEXT NOT NULL UNIQUE,"
76 "latest_release_id INTEGER DEFAULT -1,"
77 "token TEXT NOT NULL,"
78 "number_of_downloads INTEGER DEFAULT 0,"
79 "builds TEXT"
80 ")")
81 self._database.execute("CREATE TABLE IF NOT EXISTS releases("
82 "release_id INTEGER PRIMARY KEY,"
83 "package_id INTEGER,"
84 "version TEXT NOT NULL,"
85 "description TEXT NOT NULL,"
86 "UNIQUE(package_id, version)"
87 ")")
88 self._database.execute("CREATE TABLE IF NOT EXISTS dependents("
89 "name TEXT NOT NULL,"
90 "user TEXT NOT NULL,"
91 "UNIQUE(name, user)"
92 ")")
93 self._database.execute("CREATE TABLE IF NOT EXISTS activities("
94 "date TEXT NOT NULL,"
95 "kind TEXT NOT NULL,"
96 "message TEXT NOT NULL"
97 ")")
99 statement = self._database.prepare("SELECT COUNT(*) FROM packages")
100 statement.fetch()
101 print("Number of packages:", statement.column_int(0))
103 self._create_mys = self._database.prepare(
104 "INSERT INTO mys (token) VALUES(?)")
105 self._modify_mys = self._database.prepare(
106 "UPDATE mys SET latest_release_id = ?")
107 self._get_mys = self._database.prepare("SELECT * FROM mys")
108 self._get_mys_release_by_id = self._database.prepare(
109 "SELECT * FROM mys_releases WHERE release_id == ?")
110 self._add_mys_release = self._database.prepare(
111 "INSERT OR IGNORE INTO mys_releases (version) VALUES(?)")
112 self._get_mys_release = self._database.prepare(
113 "SELECT * FROM mys_releases WHERE version == ?")
114 self._get_package = self._database.prepare(
115 "SELECT * FROM packages WHERE name == ?")
116 self._get_packages = self._database.prepare(
117 "SELECT (name) FROM packages ORDER BY name ASC")
118 self._get_package_release = self._database.prepare(
119 "SELECT * FROM releases WHERE package_id == ? AND version == ?")
120 self._get_package_release_by_id = self._database.prepare(
121 "SELECT * FROM releases WHERE release_id == ?")
122 self._get_package_releases = self._database.prepare(
123 "SELECT * FROM releases WHERE package_id == ?")
124 self._create_package = self._database.prepare(
125 "INSERT INTO packages (name, token) VALUES(?, ?)")
126 self._delete_package = self._database.prepare(
127 "DELETE FROM packages WHERE name = ?")
128 self._delete_all_package_releases = self._database.prepare(
129 "DELETE FROM releases WHERE package_id = ?")
130 self._modify_package = self._database.prepare(
131 "UPDATE packages SET latest_release_id = ? WHERE name == ?")
132 self._increment_package_download_count = self._database.prepare(
133 "UPDATE packages SET number_of_downloads = number_of_downloads + 1 "
134 "WHERE name == ?")
135 self._set_package_builds = self._database.prepare(
136 "UPDATE packages SET builds = ? WHERE name == ?")
137 self._add_package_release = self._database.prepare(
138 "INSERT OR IGNORE INTO releases (package_id, version, description) "
139 "VALUES(?, ?, ?)")
140 self._get_dependents = self._database.prepare(
141 "SELECT (user) FROM dependents WHERE name == ?")
142 self._remove_dependents = self._database.prepare(
143 "DELETE FROM dependents WHERE user == ?")
144 self._add_dependent = self._database.prepare(
145 "INSERT INTO dependents (name, user) VALUES (?, ?)")
146 self._clear_activities = self._database.prepare(
147 "DELETE FROM activities")
148 self._add_activity = self._database.prepare(
149 "INSERT INTO activities (date, kind, message) VALUES(?, ?, ?)")
150 self._get_activities = self._database.prepare(
151 "SELECT * FROM activities")
153 self.make_path("package").mkdir(exists_ok=True)
155 func begin_transaction(self):
156 self._lock.acquire()
157 self._database.execute("BEGIN TRANSACTION")
159 func commit_transaction(self):
160 self._database.execute("COMMIT")
161 self._lock.release()
163 func rollback_transaction(self):
164 self._database.execute("ROLLBACK")
165 self._lock.release()
167 func make_path(self, path: string) -> Path:
168 """Prepend the database root directory path to given path. Given path
169 must not start with a slash.
171 """
173 return self.root_directory.join(path)
175 func create_package(self, name: string, token: string):
176 self._create_package.bind_string(1, name)
177 self._create_package.bind_string(2, token)
178 self._create_package.execute()
180 func delete_package(self, package: Package):
181 self._delete_package.bind_string(1, package.name)
182 self._delete_package.execute()
183 self._delete_all_package_releases.bind_int(1, package.package_id)
184 self._delete_package.execute()
186 func modify_package(self, package: Package, latest_release: Release):
187 self._modify_package.bind_int(1, latest_release.release_id)
188 self._modify_package.bind_string(2, package.name)
189 self._modify_package.execute()
191 func increment_package_download_count(self, package_name: string):
192 self._increment_package_download_count.bind_string(1, package_name)
193 self._increment_package_download_count.execute()
195 func set_package_builds(self, package_name: string, value: string):
196 self._set_package_builds.bind_string(1, value)
197 self._set_package_builds.bind_string(2, package_name)
198 self._set_package_builds.execute()
200 func get_package(self, name: string) -> Package?:
201 self._get_package.bind_string(1, name)
203 if not self._get_package.fetch():
204 return None
206 package = Package(self._get_package.column_int(0),
207 self._get_package.column_string(1),
208 None,
209 [],
210 self._get_package.column_string(3),
211 self._get_package.column_int(4),
212 self._get_package.column_string(5))
213 latest_release_id = self._get_package.column_int(2)
214 self._get_package.fetch()
216 if latest_release_id != -1:
217 self._get_package_release_by_id.bind_int(1, latest_release_id)
218 self._get_package_release_by_id.fetch()
219 package.latest_release = Release(
220 self._get_package_release_by_id.column_int(0),
221 self._get_package_release_by_id.column_string(2),
222 self._get_package_release_by_id.column_string(3))
223 self._get_package_release_by_id.fetch()
225 self._get_package_releases.bind_int(1, package.package_id)
227 while self._get_package_releases.fetch():
228 package.releases.append(Release(self._get_package_releases.column_int(1),
229 self._get_package_releases.column_string(2),
230 self._get_package_releases.column_string(3)))
232 return package
234 func get_packages(self) -> [string]:
235 packages: [string] = []
237 while self._get_packages.fetch():
238 packages.append(self._get_packages.column_string(0))
240 return packages
242 func add_package_release(self,
243 package: Package,
244 version: string,
245 description: string):
246 self._add_package_release.bind_int(1, package.package_id)
247 self._add_package_release.bind_string(2, version)
248 self._add_package_release.bind_string(3, description)
249 self._add_package_release.execute()
251 func get_package_release(self, package: Package, version: string) -> Release?:
252 self._get_package_release.bind_int(1, package.package_id)
253 self._get_package_release.bind_string(2, version)
255 if not self._get_package_release.fetch():
256 return None
258 release = Release(self._get_package_release.column_int(0),
259 self._get_package_release.column_string(2),
260 self._get_package_release.column_string(3))
261 self._get_package_release.fetch()
263 return release
265 func create_mys(self, token: string):
266 self._create_mys.bind_string(1, token)
267 self._create_mys.execute()
269 func modify_mys(self, latest_release: MysRelease):
270 self._modify_mys.bind_int(1, latest_release.release_id)
271 self._modify_mys.execute()
273 func get_mys(self) -> Mys?:
274 if not self._get_mys.fetch():
275 return None
277 latest_release_id = self._get_mys.column_int(1)
278 token = self._get_mys.column_string(2)
279 self._get_mys.fetch()
280 latest_release: MysRelease? = None
282 if latest_release_id != -1:
283 self._get_mys_release_by_id.bind_int(1, latest_release_id)
284 self._get_mys_release_by_id.fetch()
285 latest_release = MysRelease(
286 self._get_mys_release_by_id.column_int(0),
287 self._get_mys_release_by_id.column_string(1))
288 self._get_mys_release_by_id.fetch()
290 return Mys(latest_release, token)
292 func add_mys_release(self, version: string):
293 self._add_mys_release.bind_string(1, version)
294 self._add_mys_release.execute()
296 func get_mys_release(self, version: string) -> MysRelease?:
297 self._get_mys_release.bind_string(1, version)
299 if not self._get_mys_release.fetch():
300 return None
302 release = MysRelease(self._get_mys_release.column_int(0),
303 self._get_mys_release.column_string(1))
304 self._get_mys_release.fetch()
306 return release
308 func remove_dependents(self, user: string):
309 self._remove_dependents.bind_string(1, user)
310 self._remove_dependents.execute()
312 func add_dependent(self, package_name: string, user: string):
313 self._add_dependent.bind_string(1, package_name)
314 self._add_dependent.bind_string(2, user)
315 self._add_dependent.execute()
317 func get_dependents(self, package_name: string) -> [string]:
318 self._get_dependents.bind_string(1, package_name)
319 dependents: [string] = []
321 while self._get_dependents.fetch():
322 dependents.append(self._get_dependents.column_string(0))
324 return dependents
326 func clear_activities(self):
327 self._clear_activities.execute()
329 func add_activity(self, date: string, kind: string, message: string):
330 self._add_activity.bind_string(1, date)
331 self._add_activity.bind_string(2, kind)
332 self._add_activity.bind_string(3, message)
333 self._add_activity.execute()
335 func get_activities(self) -> [(string, string, string)]:
336 activities: [(string, string, string)] = []
338 while self._get_activities.fetch():
339 activities.append((self._get_activities.column_string(0),
340 self._get_activities.column_string(1),
341 self._get_activities.column_string(2)))
343 return activities