83 lines
2.1 KiB
Python
83 lines
2.1 KiB
Python
from dataclasses import dataclass, fields
|
|
|
|
from .common import NoteLinesMixin, create_email
|
|
from .connection import with_cursor
|
|
from .users import User
|
|
|
|
|
|
@dataclass
|
|
class Alias(NoteLinesMixin):
|
|
id: int
|
|
source_email_id: int
|
|
email: str
|
|
destination_user_id: int
|
|
enabled: bool
|
|
note: str
|
|
|
|
|
|
select_fields = ", ".join(
|
|
map(
|
|
lambda field: f"{'a' if field.name != 'email' else 'e'}.{field.name}",
|
|
fields(Alias),
|
|
)
|
|
)
|
|
base_alias_query = f"""\
|
|
SELECT {select_fields} FROM aliases a INNER JOIN emails e ON a.source_email_id=e.id"""
|
|
|
|
|
|
@with_cursor
|
|
def get_aliases(cur, user_id):
|
|
cur.execute(
|
|
f"{base_alias_query} WHERE a.destination_user_id=%s ORDER BY e.email ASC",
|
|
[user_id],
|
|
)
|
|
return [Alias(*row) for row in cur.fetchall()]
|
|
|
|
|
|
@with_cursor
|
|
def get_alias_with_user_by_id(cur, alias_id):
|
|
cur.execute(
|
|
"""\
|
|
SELECT a.id, a.source_email_id, ea.email, a.destination_user_id, a.enabled, a.note,
|
|
u.id, u.email_id, eu.email, u.passwordhash, u.enabled, u.is_admin,
|
|
u.can_use_different_alias_domain, u.note
|
|
FROM aliases a
|
|
INNER JOIN emails ea ON a.source_email_id=ea.id
|
|
INNER JOIN users u ON a.destination_user_id=u.id
|
|
INNER JOIN emails eu ON u.email_id=eu.id
|
|
WHERE a.id=%s""",
|
|
[alias_id],
|
|
)
|
|
row = cur.fetchone()
|
|
if not row:
|
|
return None
|
|
return Alias(*row[:6]), User(*row[6:])
|
|
|
|
|
|
@with_cursor
|
|
def create_alias(cur, email, user, enabled, note):
|
|
email_id = create_email(cur, email)
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO aliases (source_email_id, destination_user_id, enabled, note)
|
|
VALUES (%s, %s, %s, %s)""",
|
|
[email_id, user.id, enabled, note],
|
|
)
|
|
|
|
|
|
@with_cursor
|
|
def update_alias(cur, id, **fields):
|
|
"""updates enabled and note"""
|
|
cur.execute(
|
|
"UPDATE aliases SET "
|
|
+ ", ".join(f"{field}=%s" for field in fields.keys())
|
|
+ " WHERE id=%s",
|
|
list(fields.values()) + [id],
|
|
)
|
|
|
|
|
|
@with_cursor
|
|
def delete_alias(cur, alias):
|
|
cur.execute("DELETE FROM aliases WHERE id=%s", [alias.id])
|
|
cur.execute("DELETE FROM emails WHERE id=%s", [alias.source_email_id])
|