tiararodney/httpaste
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(backend/sqlite): normalize functions

Browse source
This commit is contained in:
Tiara Rodney 2022-04-03 00:29:44 +02:00
parent 6c1ffdcc0c
commit 23d5128ea7
2 changed files with 82 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

75
src/httpaste/backend/sqlite/paste.py
View file

@ -9,73 +9,92 @@ def load(proto: object, connection: Connection, model_class: type):
"""load a paste
"""
cur = connection.cursor()
cursor = connection.cursor()
cur.execute(
'SELECT pid, data, data_hash, sub, expiration, encoding FROM pastes WHERE pid=?',
(proto.pid,
))
statement = '''SELECT pid, data, data_hash, sub, expiration, encoding
FROM pastes
WHERE pid=?'''
result = cur.fetchone()
cursor.execute(statement, (proto.pid,))
if result:
row = cursor.fetchone()
if row is not None:
return model_class(
result['pid'],
result['sub'],
result['data'],
result['data_hash'],
result['expiration'],
result['encoding'])
row['pid'],
row['sub'],
row['data'],
row['data_hash'],
row['expiration'],
row['encoding'])
return None
def dump(model: object, connection: Connection):
def dump(model: object, connection: Connection) -> None:
"""dump a paste
"""
cur = connection.cursor()
cursor = connection.cursor()
cur.execute(
'''INSERT INTO pastes (pid, data, data_hash, sub, expiration, encoding)
VALUES (?,?,?,?,?,?)''',
(model.pid,
statement = '''INSERT INTO pastes
(pid, data, data_hash, sub, expiration, encoding)
VALUES (?,?,?,?,?,?)'''
values = (model.pid,
model.data,
model.data_hash,
model.sub,
model.expiration,
model.encoding))
model.encoding)
cursor.execute(statement, values)
connection.commit()
return None
def delete(proto: object, connection: Connection) -> bool:
cur = connection.cursor()
def delete(proto: object, connection: Connection) -> None:
cur.execute('''DELETE FROM pastes WHERE pid=?''', (proto.pid,))
cursor = connection.cursor()
cursor.execute('''DELETE FROM pastes WHERE pid=?''', (proto.pid,))
connection.commit()
return None
def init(connection: Connection):
cur = connection.cursor()
cursor = connection.cursor()
with open(path.join(path.dirname(__file__), 'paste.sql'), 'r') as fh:
cur.execute(fh.read())
statement = fh.read()
cursor.execute(statement)
connection.commit()
def sanitize(connection: Connection, model_class: type) -> bool:
def sanitize(connection: Connection, model_class: type) -> int:
cur = connection.cursor()
cursor = connection.cursor()
cur.execute('''SELECT pid FROM pastes WHERE expiration < ? AND expiration > 0''', (int(time()),))
statement = '''SELECT pid FROM pastes
WHERE expiration < ? AND expiration > 0'''
cursor.execute(statement, (int(time()),))
srow_count = 0
for row in cur.fetchall():
delete(model_class(row['pid']))
srow_count += 1
return srow_count

56
src/httpaste/backend/sqlite/user.py
View file

@ -2,59 +2,73 @@
"""
from os import path
from sqlite3 import Connection
from httpaste.model import User
def load(proto: User, connection: Connection):
def load(proto: object, connection: Connection, model_class: type):
"""load a user
"""
cur = connection.cursor()
cursor = connection.cursor()
cur.execute(
'SELECT sub, key_hash, paste_index FROM users WHERE sub=?', (proto.sub,))
statement = '''SELECT sub, key_hash, paste_index
FROM users
WHERE sub=?'''
result = cur.fetchone()
cursor.execute(statement, (proto.sub,))
if result:
row = cursor.fetchone()
return User(result['sub'], result['key_hash'], result['paste_index'])
if row is not None:
return model_class(result['sub'], result['key_hash'],
result['paste_index'])
return None
def dump(model: User, connection: Connection):
def dump(model: object, connection: Connection) -> None:
"""dump a user
"""
cur = connection.cursor()
cursor = connection.cursor()
cur.execute('''INSERT OR REPLACE INTO users (sub, key_hash, paste_index)
VALUES (?,?,?)''', (model.sub, model.key_hash, model.index))
statement = '''INSERT OR REPLACE INTO users
(sub, key_hash, paste_index)
VALUES (?,?,?)'''
cursor.execute(statement, (model.sub, model.key_hash, model.index))
connection.commit()
return None
def delete(proto: object, connection: Connection) -> bool:
cur = connection.cursor()
def delete(proto: object, connection: Connection) -> None:
cur.execute('''DELETE FROM users WHERE sub=?''', (proto.sub,))
cursor = connection.cursor()
cursor.execute('''DELETE FROM users WHERE sub=?''', (proto.sub,))
connection.commit()
return None
def init(connection: Connection):
cur = connection.cursor()
def init(connection: Connection) -> None:
cursor = connection.cursor()
with open(path.join(path.dirname(__file__), 'user.sql'), 'r') as fh:
cur.execute(fh.read())
statement = fh.read()
cursor.execute(statement)
connection.commit()
def sanitize(connection: Connection, model_class) -> bool:
return None
def sanitize(connection: Connection, model_class) -> int:
return 0