Renamed src/database into src/db

src/db/__init__.py

@@ -0,0 +1,3 @@
+from . import add, change, delete, get, models, prepare
+
+__all__ = ["add", "delete", "get", "models", "prepare", "change"]

src/db/add.py

@@ -0,0 +1,31 @@
+import sqlmodel
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.future import Engine
+
+from . import models
+
+
+def add_account(engine: Engine, account: models.Account) -> bool:
+    """Adds account to the database. Returns true on success,
+    false otherwise"""
+    try:
+        with sqlmodel.Session(engine) as session:
+            session.add(account)
+            session.commit()
+    except IntegrityError:
+        return False
+    else:
+        return True
+
+
+def add_master_pass(engine: Engine, master_pass: models.MasterPass) -> bool:
+    """Adds master password the database. Returns true on success,
+    false otherwise"""
+    try:
+        with sqlmodel.Session(engine) as session:
+            session.add(master_pass)
+            session.commit()
+    except IntegrityError:
+        return False
+    else:
+        return True

src/db/change.py

@@ -0,0 +1,22 @@
+import sqlmodel
+from sqlalchemy.future import Engine
+
+from . import models
+
+
+def change_master_pass(
+    engine: Engine,
+    master_password: models.MasterPass,
+) -> None:
+    """Changes master password and salt in the database"""
+    statement = (
+        sqlmodel.update(models.MasterPass)
+        .where(models.MasterPass.user_id == master_password.user_id)
+        .values(
+            salt=master_password.salt,
+            password_hash=master_password.password_hash,
+        )
+    )
+    with sqlmodel.Session(engine) as session:
+        session.exec(statement)
+        session.commit()

src/db/delete.py

@@ -0,0 +1,35 @@
+import sqlmodel
+from sqlalchemy.future import Engine
+
+from . import models
+
+
+def purge_accounts(engine: Engine, user_id: int) -> None:
+    """Deletes all user's accounts"""
+    statement = sqlmodel.delete(models.Account).where(
+        models.Account.user_id == user_id,
+    )
+    with sqlmodel.Session(engine) as session:
+        session.exec(statement)
+        session.commit()
+
+
+def delete_master_pass(engine: Engine, user_id: int) -> None:
+    """Delets master password of the user"""
+    statement = sqlmodel.delete(models.MasterPass).where(
+        models.MasterPass.user_id == user_id
+    )
+    with sqlmodel.Session(engine) as session:
+        session.exec(statement)
+        session.commit()
+
+
+def delete_account(engine: Engine, user_id: int, name: str) -> None:
+    """Deletes specific user account"""
+    statement = sqlmodel.delete(models.Account).where(
+        models.Account.user_id == user_id,
+        models.Account.name == name,
+    )
+    with sqlmodel.Session(engine) as session:
+        session.exec(statement)
+        session.commit()

src/db/get.py

@@ -0,0 +1,59 @@
+import sqlmodel
+from sqlalchemy.future import Engine
+
+from . import models
+
+
+def get_master_pass(
+    engine: Engine,
+    user_id: int,
+) -> models.MasterPass | None:
+    """Gets master password of a user"""
+    statement = sqlmodel.select(models.MasterPass).where(
+        models.MasterPass.user_id == user_id,
+    )
+    with sqlmodel.Session(engine) as session:
+        result = session.exec(statement).first()
+    return result
+
+
+def get_accounts(
+    engine: Engine,
+    user_id: int,
+    *,
+    to_sort: bool = False,
+) -> list[str]:
+    """Gets a list of account names of a user"""
+    statement = sqlmodel.select(models.Account.name).where(
+        models.Account.user_id == user_id,
+    )
+    if to_sort:
+        statement = statement.order_by(models.Account.name)
+    with sqlmodel.Session(engine) as session:
+        result = session.exec(statement).fetchall()
+    return result
+
+
+def get_all_accounts(engine: Engine, user_id: int) -> list[models.Account]:
+    """Returns a list of accounts of a user"""
+    statement = sqlmodel.select(models.Account).where(
+        models.Account.user_id == user_id,
+    )
+    with sqlmodel.Session(engine) as session:
+        result = session.exec(statement).fetchall()
+    return result
+
+
+def get_account_info(
+    engine: Engine,
+    user_id: int,
+    name: str,
+) -> models.Account:
+    """Gets account info"""
+    statement = sqlmodel.select(models.Account).where(
+        models.Account.user_id == user_id,
+        models.Account.name == name,
+    )
+    with sqlmodel.Session(engine) as session:
+        result = session.exec(statement).first()
+    return result

src/db/models.py

@@ -0,0 +1,30 @@
+import sqlmodel
+
+
+class MasterPass(sqlmodel.SQLModel, table=True):
+    __tablename__ = "master_passwords"
+    user_id: int = sqlmodel.Field(
+        sa_column=sqlmodel.Column(sqlmodel.INT(), primary_key=True)
+    )
+    salt: bytes = sqlmodel.Field(
+        sa_column=sqlmodel.Column(sqlmodel.BINARY(64), nullable=False)
+    )
+    password_hash: bytes = sqlmodel.Field(
+        sa_column=sqlmodel.Column(sqlmodel.BINARY(128), nullable=False)
+    )
+
+
+class Account(sqlmodel.SQLModel, table=True):
+    __tablename__ = "accounts"
+    __table_args__ = (sqlmodel.PrimaryKeyConstraint("user_id", "name"),)
+    user_id: int = sqlmodel.Field()
+    name: str = sqlmodel.Field(max_length=255)
+    salt: bytes = sqlmodel.Field(
+        sa_column=sqlmodel.Column(sqlmodel.BINARY(64), nullable=False)
+    )
+    enc_login: bytes = sqlmodel.Field(
+        sa_column=sqlmodel.Column(sqlmodel.VARBINARY(256), nullable=False)
+    )
+    enc_password: bytes = sqlmodel.Field(
+        sa_column=sqlmodel.Column(sqlmodel.VARBINARY(256), nullable=False)
+    )

src/db/prepare.py

@@ -0,0 +1,18 @@
+import sqlmodel
+from sqlalchemy.future import Engine
+
+from . import models  # noqa
+
+HOUR_IN_SECONDS = 3600
+
+
+def get_engine(host: str, user: str, passwd: str, db: str) -> Engine:
+    """Creates an engine for mariadb with pymysql as connector"""
+    uri = f"mariadb+pymysql://{user}:{passwd}@{host}/{db}"
+    engine = sqlmodel.create_engine(uri, pool_recycle=HOUR_IN_SECONDS)
+    return engine
+
+
+def prepare(engine: Engine) -> None:
+    """Creates all tables, indexes and constrains in the database"""
+    sqlmodel.SQLModel.metadata.create_all(engine)