Fix more pylint warnings

This commit is contained in:
augustin64 2023-12-15 13:38:32 +01:00
parent 11ddbf0169
commit acaa8367d6
13 changed files with 189 additions and 147 deletions

5
.pylintrc Normal file
View File

@ -0,0 +1,5 @@
[BASIC]
good-names=db, v, v1, v2, f, id, i, j, k
[MESSAGES CONTROL]
disable=pointless-string-statement

View File

@ -57,26 +57,10 @@ def add_user():
username = request.form["username"] username = request.form["username"]
password = request.form["password"] password = request.form["password"]
album_uuid = request.form["album_uuid"] album_uuid = request.form["album_uuid"]
db = get_db()
error = None
if not username: error = auth.create_user(username, password)
error = "Un nom d'utilisateur est requis."
elif not password:
error = "Un mot de passe est requis."
if error is None: if error is None:
try:
db.execute(
"INSERT INTO user (username, password) VALUES (?, ?)",
(username, generate_password_hash(password)),
)
db.commit()
except db.IntegrityError:
# The username was already taken, which caused the
# commit to fail. Show a validation error.
error = f"Le nom d'utilisateur {username} est déjà pris."
else:
# Success, go to the login page. # Success, go to the login page.
user = User(name=username) user = User(name=username)
try: try:

View File

@ -25,10 +25,10 @@ def index():
SELECT id FROM user SELECT id FROM user
""" """
) )
users = [User(user_id=u["id"]) for u in users_id] users = [User(user_id=user["id"]) for user in users_id]
for u in users: for user in users:
u.albums = u.get_albums() user.get_albums()
u.partitions = u.get_partitions() user.get_partitions()
return render_template( return render_template(
"admin/index.html", "admin/index.html",

View File

@ -65,7 +65,7 @@ def search_page():
) )
@bp.route("/<uuid>") @bp.route("/<uuid>")
def album(uuid): def get_album(uuid):
""" """
Album page Album page
""" """

View File

@ -3,6 +3,7 @@
Authentification module Authentification module
""" """
import functools import functools
from typing import Optional
from flask import (Blueprint, flash, g, redirect, render_template, from flask import (Blueprint, flash, g, redirect, render_template,
request, session, url_for, current_app) request, session, url_for, current_app)
@ -75,6 +76,30 @@ def load_logged_in_user():
) )
def create_user(username: str, password: str) -> Optional[str]:
"""Adds a new user to the database"""
if not username:
error = "Un nom d'utilisateur est requis."
elif not password:
error = "Un mot de passe est requis."
try:
db = get_db()
db.execute(
"INSERT INTO user (username, password) VALUES (?, ?)",
(username, generate_password_hash(password)),
)
db.commit()
except db.IntegrityError:
# The username was already taken, which caused the
# commit to fail. Show a validation error.
error = f"Le nom d'utilisateur {username} est déjà pris."
if error is not None:
return error
@bp.route("/register", methods=("GET", "POST")) @bp.route("/register", methods=("GET", "POST"))
@anon_required @anon_required
def register(): def register():
@ -89,32 +114,13 @@ def register():
if request.method == "POST": if request.method == "POST":
username = request.form["username"] username = request.form["username"]
password = request.form["password"] password = request.form["password"]
db = get_db()
error = None
if not username: error = create_user(username, password)
error = "Un nom d'utilisateur est requis."
elif not password:
error = "Un mot de passe est requis."
if error is None:
try:
db.execute(
"INSERT INTO user (username, password) VALUES (?, ?)",
(username, generate_password_hash(password)),
)
db.commit()
flash(f"Utilisateur {username} créé avec succès. Vous pouvez vous connecter.")
except db.IntegrityError:
# The username was already taken, which caused the
# commit to fail. Show a validation error.
error = f"Le nom d'utilisateur {username} est déjà pris. \
Vous souhaitez peut-être vous connecter"
else:
# Success, go to the login page.
return redirect(url_for("auth.login"))
if error is not None:
flash(error) flash(error)
else:
flash("Utilisateur créé avec succès. Vous pouvez vous connecter.")
return render_template("auth/register.html") return render_template("auth/register.html")

View File

@ -60,7 +60,7 @@ class Partition():
def update(self, name=None, author="", body=""): def update(self, name=None, author="", body=""):
if name is None: if name is None:
raise Exception("name cannot be None") raise ValueError("name cannot be None")
db = get_db() db = get_db()
db.execute( db.execute(

View File

@ -18,7 +18,7 @@ def index():
@bp.route("/<uuid>") @bp.route("/<uuid>")
def groupe(uuid): def get_groupe(uuid):
""" """
Groupe page Groupe page
""" """
@ -221,7 +221,7 @@ def create_album(groupe_uuid):
@bp.route("/<groupe_uuid>/<album_uuid>") @bp.route("/<groupe_uuid>/<album_uuid>")
def album(groupe_uuid, album_uuid): def get_album(groupe_uuid, album_uuid):
""" """
Album page Album page
""" """
@ -246,7 +246,7 @@ def album(groupe_uuid, album_uuid):
user = User(user_id=session.get("user_id")) user = User(user_id=session.get("user_id"))
# List of users without duplicate # List of users without duplicate
users_id = list(set([i["id"] for i in album.get_users()+groupe.get_users()])) users_id = list({i["id"] for i in album.get_users()+groupe.get_users()})
album.users = [User(user_id=id) for id in users_id] album.users = [User(user_id=id) for id in users_id]
partitions = album.get_partitions() partitions = album.get_partitions()

View File

@ -14,7 +14,7 @@ from .utils import get_all_partitions, User, Partition, Attachment
bp = Blueprint("partition", __name__, url_prefix="/partition") bp = Blueprint("partition", __name__, url_prefix="/partition")
@bp.route("/<uuid>") @bp.route("/<uuid>")
def partition(uuid): def get_partition(uuid):
try: try:
partition = Partition(uuid=uuid) partition = Partition(uuid=uuid)
except LookupError: except LookupError:
@ -107,7 +107,7 @@ def add_attachment(uuid):
@bp.route("/attachment/<uuid>.<filetype>") @bp.route("/attachment/<uuid>.<filetype>")
def attachment(uuid, filetype): def get_attachment(uuid, filetype):
try: try:
attachment = Attachment(uuid=uuid) attachment = Attachment(uuid=uuid)
except LookupError: except LookupError:

View File

@ -57,7 +57,7 @@ def download_search_result(element):
urllib.request.urlretrieve(url, f"partitioncloud/search-partitions/{uuid}.pdf") urllib.request.urlretrieve(url, f"partitioncloud/search-partitions/{uuid}.pdf")
except (urllib.error.HTTPError, urllib.error.URLError): except (urllib.error.HTTPError, urllib.error.URLError):
with open(f"partitioncloud/search-partitions/{uuid}.pdf", 'a', encoding="utf8") as f: with open(f"partitioncloud/search-partitions/{uuid}.pdf", 'a', encoding="utf8") as _:
pass # Create empty file pass # Create empty file

View File

@ -4,8 +4,8 @@ import random
import string import string
import qrcode import qrcode
from .db import get_db
from flask import current_app, send_file from flask import current_app, send_file
from .db import get_db
def new_uuid(): def new_uuid():

View File

@ -2,6 +2,7 @@ import random
import string import string
import sqlite3 import sqlite3
def run_sqlite_command(*args): def run_sqlite_command(*args):
"""Run a command against the database""" """Run a command against the database"""
con = sqlite3.connect("instance/partitioncloud.sqlite") con = sqlite3.connect("instance/partitioncloud.sqlite")
@ -10,6 +11,7 @@ def run_sqlite_command(*args):
con.commit() con.commit()
con.close() con.close()
def get_sqlite_data(*args): def get_sqlite_data(*args):
"""Get data from the db""" """Get data from the db"""
con = sqlite3.connect("instance/partitioncloud.sqlite") con = sqlite3.connect("instance/partitioncloud.sqlite")
@ -19,8 +21,12 @@ def get_sqlite_data(*args):
con.close() con.close()
return new_data return new_data
def new_uuid(): def new_uuid():
return ''.join([random.choice(string.ascii_uppercase + string.digits) for _ in range(6)]) return "".join(
[random.choice(string.ascii_uppercase + string.digits) for _ in range(6)]
)
def format_uuid(uuid): def format_uuid(uuid):
"""Format old uuid4 format""" """Format old uuid4 format"""

View File

@ -6,10 +6,11 @@ from colorama import Fore, Style
""" """
v1.3.* v1.3.*
""" """
def add_source(): def add_source():
utils.run_sqlite_command( utils.run_sqlite_command("ALTER TABLE partition ADD source TEXT DEFAULT 'unknown'")
"ALTER TABLE partition ADD source TEXT DEFAULT 'unknown'"
)
def add_groupes(): def add_groupes():
utils.run_sqlite_command( utils.run_sqlite_command(
@ -35,6 +36,7 @@ def add_groupes():
);""" );"""
) )
def add_attachments(): def add_attachments():
os.makedirs("partitioncloud/attachments", exist_ok=True) os.makedirs("partitioncloud/attachments", exist_ok=True)
utils.run_sqlite_command( utils.run_sqlite_command(
@ -47,6 +49,7 @@ def add_attachments():
);""" );"""
) )
def install_colorama(): def install_colorama():
os.system("pip install colorama -qq") os.system("pip install colorama -qq")
@ -54,17 +57,15 @@ def install_colorama():
""" """
v1.4.* v1.4.*
""" """
def mass_rename(): def mass_rename():
"""Rename all albums & groupes to use a shorter uuid""" """Rename all albums & groupes to use a shorter uuid"""
albums = utils.get_sqlite_data("SELECT * FROM album") albums = utils.get_sqlite_data("SELECT * FROM album")
groupes = utils.get_sqlite_data("SELECT * FROM groupe") groupes = utils.get_sqlite_data("SELECT * FROM groupe")
utils.run_sqlite_command( utils.run_sqlite_command("ALTER TABLE groupe RENAME TO _groupe_old")
"ALTER TABLE groupe RENAME TO _groupe_old" utils.run_sqlite_command("ALTER TABLE album RENAME TO _album_old")
)
utils.run_sqlite_command(
"ALTER TABLE album RENAME TO _album_old"
)
utils.run_sqlite_command( # Add UNIQUE constraint & change uuid length utils.run_sqlite_command( # Add UNIQUE constraint & change uuid length
"""CREATE TABLE groupe ( """CREATE TABLE groupe (
@ -89,17 +90,20 @@ def mass_rename():
INSERT INTO album (id, name, uuid) INSERT INTO album (id, name, uuid)
VALUES (?, ?, ?) VALUES (?, ?, ?)
""", """,
(album[0], album[1], utils.format_uuid(album[2])) (album[0], album[1], utils.format_uuid(album[2])),
) )
except sqlite3.IntegrityError: except sqlite3.IntegrityError:
uuid = utils.new_uuid() uuid = utils.new_uuid()
print(f"{Fore.RED}Collision on {album[1]}{Style.RESET_ALL} ({album[2][:10]} renaming to {uuid})") print(
f"{Fore.RED}Collision on {album[1]}{Style.RESET_ALL} \
({album[2][:10]} renaming to {uuid})"
)
utils.run_sqlite_command( utils.run_sqlite_command(
""" """
INSERT INTO album (id, name, uuid) INSERT INTO album (id, name, uuid)
VALUES (?, ?, ?) VALUES (?, ?, ?)
""", """,
(album[0], album[1], uuid) (album[0], album[1], uuid),
) )
for groupe in groupes: for groupe in groupes:
@ -109,28 +113,32 @@ def mass_rename():
INSERT INTO groupe (id, name, uuid) INSERT INTO groupe (id, name, uuid)
VALUES (?, ?, ?) VALUES (?, ?, ?)
""", """,
(groupe[0], groupe[1], utils.format_uuid(groupe[2])) (groupe[0], groupe[1], utils.format_uuid(groupe[2])),
) )
except sqlite3.IntegrityError: except sqlite3.IntegrityError:
uuid = utils.new_uuid() uuid = utils.new_uuid()
print(f"{Fore.RED}Collision on {groupe[1]}{Style.RESET_ALL} ({groupe[2][:10]} renaming to {uuid})") print(
f"{Fore.RED}Collision on {groupe[1]}{Style.RESET_ALL} \
({groupe[2][:10]} renaming to {uuid})"
)
utils.run_sqlite_command( utils.run_sqlite_command(
""" """
INSERT INTO groupe (id, name, uuid) INSERT INTO groupe (id, name, uuid)
VALUES (?, ?, ?) VALUES (?, ?, ?)
""", """,
(groupe[0], groupe[1], uuid) (groupe[0], groupe[1], uuid),
) )
utils.run_sqlite_command( utils.run_sqlite_command("DROP TABLE _groupe_old")
"DROP TABLE _groupe_old" utils.run_sqlite_command("DROP TABLE _album_old")
)
utils.run_sqlite_command(
"DROP TABLE _album_old"
)
def base_url_parameter_added(): def base_url_parameter_added():
print(f"{Style.BRIGHT}{Fore.YELLOW}The parameter BASE_URL has been added, reference your front url in it{Style.RESET_ALL}") print(
f"{Style.BRIGHT}{Fore.YELLOW}The parameter BASE_URL has been added, \
reference your front url in it{Style.RESET_ALL}"
)
def install_qrcode(): def install_qrcode():
os.system("pip install qrcode -qq") os.system("pip install qrcode -qq")

View File

@ -1,18 +1,18 @@
#!/usr/bin/python3 #!/usr/bin/python3
import os import os
import sys
import shutil import shutil
import argparse import argparse
from functools import cmp_to_key from functools import cmp_to_key
from distutils.dir_util import copy_tree
from colorama import Fore, Style from colorama import Fore, Style
from hooks import v1 from hooks import v1 as v1_hooks
def get_version(v: str) -> (int, int, int): def get_version(v: str) -> (int, int, int):
"""Returns a tuple (major, minor, patch from the string v{major}.{minor}.{patch})""" """Returns a tuple (major, minor, patch from the string v{major}.{minor}.{patch})"""
assert v[0] == 'v' # Check if the version is correctly formatted assert v[0] == "v" # Check if the version is correctly formatted
return tuple(map(int, v[1:].split('.'))) return tuple(map(int, v[1:].split(".")))
def is_newer(v1: str, v2: str) -> bool: def is_newer(v1: str, v2: str) -> bool:
@ -21,30 +21,24 @@ def is_newer(v1: str, v2: str) -> bool:
hooks = [ hooks = [
("v1.3.0", [ ("v1.3.0", [("add SOURCE column", v1_hooks.add_source)]),
("add SOURCE column", v1.add_source) ("v1.2.0", [("create groupe structure", v1_hooks.add_groupes)]),
]), ("v1.3.0", [("create attachment table", v1_hooks.add_attachments)]),
("v1.2.0", [ ("v1.3.3", [("Install colorama", v1_hooks.install_colorama)]),
("create groupe structure", v1.add_groupes) (
]), "v1.4.0",
("v1.3.0", [ [
("create attachment table", v1.add_attachments) ("Change all albums & groupes uuids", v1_hooks.mass_rename),
]), ("Warn new parameter", v1_hooks.base_url_parameter_added),
("v1.3.3", [ ],
("Install colorama", v1.install_colorama) ),
]), ("v1.4.1", [("Install qrcode", v1_hooks.install_qrcode)]),
("v1.4.0", [
("Change all albums & groupes uuids", v1.mass_rename),
("Warn new parameter", v1.base_url_parameter_added)
]),
("v1.4.1", [
("Install qrcode", v1.install_qrcode)
]),
] ]
def get_hooks(current, target): def get_hooks(current, target):
"""Returns a list of hooks needed to migrate""" """Returns a list of hooks needed to migrate"""
def compare(v1: str, v2: str): def compare(v1: str, v2: str):
if is_newer(v2[0], v1[0]): if is_newer(v2[0], v1[0]):
return -1 return -1
@ -54,7 +48,9 @@ def get_hooks(current, target):
applied_hooks = [] applied_hooks = []
for hook in hooks: for hook in hooks:
if is_newer(hook[0], current) and (target == hook[0] or is_newer(target, hook[0])): if is_newer(hook[0], current) and (
target == hook[0] or is_newer(target, hook[0])
):
applied_hooks.append(hook) applied_hooks.append(hook)
return sorted(applied_hooks, key=cmp_to_key(compare)) return sorted(applied_hooks, key=cmp_to_key(compare))
@ -62,43 +58,53 @@ def get_hooks(current, target):
def backup_instance(version, verbose=True): def backup_instance(version, verbose=True):
"""Backs up current instance in backups/{version}""" """Backs up current instance in backups/{version}"""
def print_verbose(*args):
def print_verbose(*f_args):
if verbose: if verbose:
print(*args) print(*f_args)
print_verbose("\nBacking up current instance") print_verbose("\nBacking up current instance")
dest = os.path.join("backups", version) dest = os.path.join("backups", version)
if os.path.exists(dest): if os.path.exists(dest):
print(f"{Fore.RED}Backup directory already exists{Style.RESET_ALL}") print(f"{Fore.RED}Backup directory already exists{Style.RESET_ALL}")
exit(1) sys.exit(1)
os.makedirs(dest) os.makedirs(dest)
paths = [ paths = [
("instance", os.path.join(dest, "instance")), ("instance", os.path.join(dest, "instance")),
(os.path.join("partitioncloud", "partitions"), os.path.join(dest, "partitions")), (
(os.path.join("partitioncloud", "attachments"), os.path.join(dest, "attachments")), os.path.join("partitioncloud", "partitions"),
(os.path.join("partitioncloud", "search-partitions"), os.path.join(dest, "search-partitions")) os.path.join(dest, "partitions"),
),
(
os.path.join("partitioncloud", "attachments"),
os.path.join(dest, "attachments"),
),
(
os.path.join("partitioncloud", "search-partitions"),
os.path.join(dest, "search-partitions"),
),
] ]
for src, dst in paths: for src, dst in paths:
if os.path.exists(src): if os.path.exists(src):
print_verbose(f"\tBacking up {src}") print_verbose(f"\tBacking up {src}")
copy_tree(src, dst) shutil.copy_tree(src, dst)
def print_hooks(hooks): def print_hooks(hooks_list):
for hook in hooks: for hook in hooks_list:
print(f"- {Fore.BLUE}{hook[0]}{Style.RESET_ALL}:") print(f"- {Fore.BLUE}{hook[0]}{Style.RESET_ALL}:")
for subhook in hook[1]: for sub_hook in hook[1]:
print("\t", subhook[0]) print("\t", sub_hook[0])
def apply_hooks(hooks): def apply_hooks(hooks_list):
for hook in hooks: for hook in hooks_list:
print(f"Migrating to {Fore.BLUE}{hook[0]}{Style.RESET_ALL}:") print(f"Migrating to {Fore.BLUE}{hook[0]}{Style.RESET_ALL}:")
for subhook in hook[1]: for sub_hook in hook[1]:
print(f"\tApplying '{subhook[0]}'") print(f"\tApplying '{sub_hook[0]}'")
subhook[1]() sub_hook[1]()
def migrate(current, target, skip_backup=False, prog_name="scripts/migration.py"): def migrate(current, target, skip_backup=False, prog_name="scripts/migration.py"):
@ -110,19 +116,24 @@ def migrate(current, target, skip_backup=False, prog_name="scripts/migration.py"
applied_hooks = get_hooks(current, target) applied_hooks = get_hooks(current, target)
if len(applied_hooks) == 0: if len(applied_hooks) == 0:
print(f"{Fore.GREEN}No hook to apply{Style.RESET_ALL}") print(f"{Fore.GREEN}No hook to apply{Style.RESET_ALL}")
exit(0) sys.exit(0)
print("The following hooks will be applied:") print("The following hooks will be applied:")
print_hooks(applied_hooks) print_hooks(applied_hooks)
if input("Apply these hooks ? [y/N] ") != "y": if input("Apply these hooks ? [y/N] ") != "y":
print(f"{Fore.RED}Aborting !{Style.RESET_ALL}") print(f"{Fore.RED}Aborting !{Style.RESET_ALL}")
exit(1) sys.exit(1)
if not skip_backup: if not skip_backup:
backup_instance(current) backup_instance(current)
print(f"Instance backed up in {Style.BRIGHT}backups/{current}{Style.RESET_ALL}\n") print(
print(f"If something goes wrong, recover with {Style.BRIGHT}{Fore.BLUE}{prog_name} --restore {current}{Style.RESET_ALL}") f"Instance backed up in {Style.BRIGHT}backups/{current}{Style.RESET_ALL}\n"
)
print(
f"If something goes wrong, recover with {Style.BRIGHT}{Fore.BLUE}{prog_name}\
--restore {current}{Style.RESET_ALL}"
)
else: else:
print("Skipping automatic backup") print("Skipping automatic backup")
@ -131,17 +142,31 @@ def migrate(current, target, skip_backup=False, prog_name="scripts/migration.py"
def restore(version): def restore(version):
if input("Do you really want to restore from backup ? Your current data will be deleted [y/N] ") != "y": if (
input(
"Do you really want to restore from backup ? Your current data will be deleted [y/N] "
)
!= "y"
):
print(f"{Fore.RED}Aborting !{Style.RESET_ALL}") print(f"{Fore.RED}Aborting !{Style.RESET_ALL}")
exit(1) sys.exit(1)
dest = os.path.join("backups", version) dest = os.path.join("backups", version)
print(f"Restoring from {dest}") print(f"Restoring from {dest}")
paths = [ paths = [
("instance", os.path.join(dest, "instance")), ("instance", os.path.join(dest, "instance")),
(os.path.join("partitioncloud", "partitions"), os.path.join(dest, "partitions")), (
(os.path.join("partitioncloud", "attachments"), os.path.join(dest, "attachments")), os.path.join("partitioncloud", "partitions"),
(os.path.join("partitioncloud", "search-partitions"), os.path.join(dest, "search-partitions")) os.path.join(dest, "partitions"),
),
(
os.path.join("partitioncloud", "attachments"),
os.path.join(dest, "attachments"),
),
(
os.path.join("partitioncloud", "search-partitions"),
os.path.join(dest, "search-partitions"),
),
] ]
for src, dst in paths: for src, dst in paths:
if os.path.exists(src): if os.path.exists(src):
@ -149,20 +174,28 @@ def restore(version):
if os.path.exists(dst): if os.path.exists(dst):
print(f"\tRestoring {src}") print(f"\tRestoring {src}")
copy_tree(dst, src) shutil.copy_tree(dst, src)
else: else:
print(f"\t{Fore.RED}No available backup for {src}, deleting current content to avoid any conflict{Style.RESET_ALL}") print(
f"\t{Fore.RED}No available backup for {src}, \
deleting current content to avoid any conflict{Style.RESET_ALL}"
)
if __name__ == "__main__": if __name__ == "__main__":
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog='PartitionCloud Migration tool', prog="PartitionCloud Migration tool",
description='Helps you migrate from one version to another') description="Helps you migrate from one version to another",
)
parser.add_argument('-c', '--current', help="current version (vx.y.z)") parser.add_argument("-c", "--current", help="current version (vx.y.z)")
parser.add_argument('-t', '--target', help="target version (vx.y.z)") parser.add_argument("-t", "--target", help="target version (vx.y.z)")
parser.add_argument('-s', '--skip-backup', action='store_true') parser.add_argument("-s", "--skip-backup", action="store_true")
parser.add_argument('-r', '--restore', help='restore from specific version backup, will not apply any hook (vx.y.z)') parser.add_argument(
"-r",
"--restore",
help="restore from specific version backup, will not apply any hook (vx.y.z)",
)
args = parser.parse_args() args = parser.parse_args()