117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
"""Email alias management."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models import Account, MailAlias, MailGroup
|
|
from app.schemas import MailAliasCreate, MailAliasRead
|
|
|
|
|
|
def _alias_read(alias: MailAlias) -> MailAliasRead:
|
|
return MailAliasRead(
|
|
id=alias.id,
|
|
alias_address=alias.alias_address,
|
|
forwards_to=alias.forwards_to,
|
|
alias_type=alias.alias_type,
|
|
is_domain_alias=alias.is_domain_alias,
|
|
status=alias.status,
|
|
created_at=alias.created_at,
|
|
)
|
|
|
|
|
|
async def list_mail_aliases(
|
|
session: AsyncSession,
|
|
company_domain: str,
|
|
) -> tuple[list[MailAliasRead], dict[str, int]]:
|
|
result = await session.scalars(
|
|
select(MailAlias)
|
|
.where(MailAlias.company_domain == company_domain)
|
|
.order_by(MailAlias.created_at.desc())
|
|
)
|
|
aliases = list(result.all())
|
|
items = [_alias_read(alias) for alias in aliases]
|
|
counts = {
|
|
"all": len(items),
|
|
"domain": sum(1 for item in items if item.is_domain_alias),
|
|
}
|
|
return items, counts
|
|
|
|
|
|
async def _destination_exists(
|
|
session: AsyncSession,
|
|
company_domain: str,
|
|
address: str,
|
|
) -> bool:
|
|
normalized = address.lower()
|
|
if not normalized.endswith(f"@{company_domain}"):
|
|
return False
|
|
|
|
account = await session.scalar(
|
|
select(Account).where(func.lower(Account.email) == normalized)
|
|
)
|
|
if account:
|
|
return True
|
|
|
|
group = await session.scalar(
|
|
select(MailGroup).where(func.lower(MailGroup.email) == normalized)
|
|
)
|
|
return group is not None
|
|
|
|
|
|
async def create_mail_alias(
|
|
session: AsyncSession,
|
|
*,
|
|
manager: Account,
|
|
payload: MailAliasCreate,
|
|
) -> MailAliasRead:
|
|
company_domain = manager.company_domain
|
|
if company_domain is None:
|
|
raise ValueError("Company domain is required.")
|
|
|
|
alias_address = str(payload.alias_address).lower()
|
|
forwards_to = str(payload.forwards_to).lower()
|
|
|
|
if not alias_address.endswith(f"@{company_domain}"):
|
|
raise ValueError(f"Alias address must use @{company_domain}.")
|
|
if not forwards_to.endswith(f"@{company_domain}"):
|
|
raise ValueError(f"Forwarding address must use @{company_domain}.")
|
|
if alias_address == forwards_to:
|
|
raise ValueError("Alias cannot forward to itself.")
|
|
|
|
existing = await session.scalar(
|
|
select(MailAlias).where(
|
|
func.lower(MailAlias.alias_address) == alias_address
|
|
)
|
|
)
|
|
if existing:
|
|
raise ValueError("That alias address is already in use.")
|
|
|
|
account_conflict = await session.scalar(
|
|
select(Account).where(func.lower(Account.email) == alias_address)
|
|
)
|
|
if account_conflict:
|
|
raise ValueError("That address is already a user mailbox.")
|
|
|
|
group_conflict = await session.scalar(
|
|
select(MailGroup).where(func.lower(MailGroup.email) == alias_address)
|
|
)
|
|
if group_conflict:
|
|
raise ValueError("That address is already a group mailbox.")
|
|
|
|
if not await _destination_exists(session, company_domain, forwards_to):
|
|
raise ValueError("Forwarding address must be an existing user or group.")
|
|
|
|
alias = MailAlias(
|
|
company_domain=company_domain,
|
|
alias_address=alias_address,
|
|
forwards_to=forwards_to,
|
|
alias_type=payload.alias_type,
|
|
is_domain_alias=payload.is_domain_alias,
|
|
status="active",
|
|
created_by_id=manager.id,
|
|
)
|
|
session.add(alias)
|
|
await session.flush()
|
|
return _alias_read(alias)
|