email_server: aliases: Minor refactoring to DB schema

- Use the IntergrityError exception instead of a complex query to ignore an
already existing alias.

- When retrieving existing aliases, use explicit list of columns instead of * so
that schema updates adding columns won't fail the code using the row results.

- Use terminology used by post fix. "name" for the name of the alias. "value"
for the mapping.

Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org>
Reviewed-by: James Valleroy <jvalleroy@mailbox.org>

plinth/modules/email_server/aliases/__init__.py

@@ -1,5 +1,5 @@
-"""Manages email aliases"""
 # SPDX-License-Identifier: AGPL-3.0-or-later
+"""Manage email aliases."""
 
 import contextlib
 import pwd
@@ -24,65 +24,62 @@ def _get_cursor():
         connection.close()
 
 
-def get(uid_number):
+def get(uid):
     """Get all aliases of a user."""
-    query = 'SELECT * FROM Alias WHERE uid_number=?'
+    query = 'SELECT name, value, status FROM alias WHERE value=?'
     with _get_cursor() as cursor:
-        rows = cursor.execute(query, (uid_number, ))
-        result = [models.Alias(**row) for row in rows]
-        return result
+        rows = cursor.execute(query, (uid, ))
+        return [models.Alias(**row) for row in rows]
 
 
-def exists(email_name):
+def exists(name):
     """Return whether alias is already taken."""
     try:
-        pwd.getpwnam(email_name)
+        pwd.getpwnam(name)
         return True
     except KeyError:
         pass
 
     with _get_cursor() as cursor:
-        query = 'SELECT COUNT(*) FROM Alias WHERE email_name=?'
-        cursor.execute(query, (email_name, ))
+        query = 'SELECT COUNT(*) FROM alias WHERE name=?'
+        cursor.execute(query, (name, ))
         return cursor.fetchone()[0] != 0
 
 
-def put(uid_number, email_name):
+def put(uid, name):
     """Insert if not exists a new alias."""
-    query = '''
-INSERT INTO Alias(email_name, uid_number, status)
-    SELECT ?,?,? WHERE NOT EXISTS (
-        SELECT 1 FROM Alias WHERE email_name=?
-    )
-'''
+    query = 'INSERT INTO alias (name, value, status) VALUES (?, ?, ?)'
     with _get_cursor() as cursor:
-        cursor.execute(query, (email_name, uid_number, 1, email_name))
+        try:
+            cursor.execute(query, (name, uid, 1))
+        except sqlite3.IntegrityError:
+            pass  # Alias exists, rare since we are already checking
 
 
-def delete(uid_number, alias_list):
+def delete(uid, aliases):
     """Delete a set of aliases."""
-    query = 'DELETE FROM Alias WHERE uid_number=? AND email_name=?'
-    parameter_seq = ((uid_number, alias) for alias in alias_list)
+    query = 'DELETE FROM alias WHERE value=? AND name=?'
+    parameter_seq = ((uid, name) for name in aliases)
     with _get_cursor() as cursor:
         cursor.execute('BEGIN')
         cursor.executemany(query, parameter_seq)
         cursor.execute('COMMIT')
 
 
-def enable(uid_number, alias_list):
+def enable(uid, aliases):
     """Enable a list of aliases."""
-    return _set_status(uid_number, alias_list, 1)
+    return _set_status(uid, aliases, 1)
 
 
-def disable(uid_number, alias_list):
+def disable(uid, aliases):
     """Disable a list of aliases."""
-    return _set_status(uid_number, alias_list, 0)
+    return _set_status(uid, aliases, 0)
 
 
-def _set_status(uid_number, alias_list, status):
+def _set_status(uid, aliases, status):
     """Set the status value of a list of aliases."""
-    query = 'UPDATE Alias SET status=? WHERE uid_number=? AND email_name=?'
-    parameter_seq = ((status, uid_number, alias) for alias in alias_list)
+    query = 'UPDATE alias SET status=? WHERE value=? AND name=?'
+    parameter_seq = ((status, uid, name) for name in aliases)
     with _get_cursor() as cursor:
         cursor.execute('BEGIN')
         cursor.executemany(query, parameter_seq)
@@ -97,11 +94,11 @@ def first_setup():
     query = '''
 PRAGMA journal_mode=WAL;
 BEGIN;
-CREATE TABLE IF NOT EXISTS Alias (
-    email_name TEXT NOT NULL,
-    uid_number INTEGER NOT NULL,
+CREATE TABLE IF NOT EXISTS alias (
+    name TEXT NOT NULL,
+    value INTEGER NOT NULL,
     status INTEGER NOT NULL,
-    PRIMARY KEY (email_name)
+    PRIMARY KEY (name)
 );
 COMMIT;
 '''

plinth/modules/email_server/aliases/models.py

@@ -3,8 +3,8 @@ from dataclasses import InitVar, dataclass, field
 
 @dataclass
 class Alias:
-    uid_number: int
-    email_name: str
+    value: int
+    name: str
     enabled: bool = field(init=False)
     status: InitVar[int]
 

plinth/modules/email_server/data/etc/postfix/freedombox-aliases.cf

@@ -2,4 +2,4 @@
 
 # Maintained by FreedomBox, do not edit.
 dbpath = /var/lib/postfix/freedombox-aliases/aliases.sqlite3
-query = SELECT uid_number FROM Alias WHERE email_name='%s' AND status=1
+query = SELECT value FROM alias WHERE name='%s' AND status=1

plinth/modules/email_server/forms.py

@@ -49,10 +49,10 @@ class AliasListForm(forms.Form):
     def __init__(self, aliases, *args, **kwargs):
         """Populate the choices for aliases."""
         super().__init__(*args, **kwargs)
-        enabled_aliases = [(alias.email_name, alias.email_name)
-                           for alias in aliases if alias.enabled]
-        disabled_aliases = [(alias.email_name, alias.email_name)
-                            for alias in aliases if not alias.enabled]
+        enabled_aliases = [(alias.name, alias.name) for alias in aliases
+                           if alias.enabled]
+        disabled_aliases = [(alias.name, alias.name) for alias in aliases
+                            if not alias.enabled]
         choices = []
         if enabled_aliases:
             choices.append((_('Enabled'), enabled_aliases))