agilentics / boiler 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
"""Organisations - the tenant boundary itself.

A worked example of the core's contract, and the shortest one in the codebase:
every function takes `session` first and `principal` second, checks a scope
before it does anything, raises a domain error rather than an HTTP status, and
returns model objects rather than JSON.
"""
import re

from db.db_declarations import Membership, Organization, User

from . import auth as auth_core
from .context import Conflict, Invalid, NotFound, Principal, Scope

_SLUG = re.compile(r"^[a-z0-9][a-z0-9-]{1,58}[a-z0-9]$")


def create_org(session, principal: Principal, name: str, slug: str) -> Organization:
    """Create a tenant. A system caller only: an organisation is created by the
    bootstrap command or by a sign-up flow you write, not by an ordinary
    signed-in user acting on some other tenant's behalf."""
    principal.require(Scope.ADMIN)

    slug = (slug or "").strip().lower()
    if not _SLUG.match(slug):
        raise Invalid("A slug is 3-60 characters of lowercase letters, digits and "
                      "hyphens, and cannot start or end with a hyphen.")
    if session.query(Organization).filter(Organization.slug == slug).one_or_none():
        raise Conflict(f"The slug {slug!r} is taken.")

    org = Organization(name=(name or slug).strip(), slug=slug)
    session.add(org)
    session.commit()
    return org


def get_org(session, principal: Principal, org_id: int) -> Organization:
    """One organisation, if this caller may see it.

    NotFound rather than Forbidden when it is another tenant's: saying "that
    exists but is not yours" confirms the existence of a customer to someone who
    should not know they have one.
    """
    principal.require(Scope.READ)
    if not principal.spans_orgs and org_id != principal.org_id:
        raise NotFound("No such organisation.")

    org = session.get(Organization, org_id)
    if org is None:
        raise NotFound("No such organisation.")
    return org


def _slugify(name: str) -> str:
    return re.sub(r"[^a-z0-9]+", "-", (name or "").lower()).strip("-")[:52]


def _unique_slug(session, *candidates: str) -> str:
    """A free slug, from the first candidate that yields one.

    Needed because the paths that create a tenant without asking for one -
    Google sign-in, any self-service sign-up you add - have a person's display
    name and nothing else. A chain rather than a single value because a display
    name is not required to contain a single slug-able character: "!!!" is
    perfectly truthy and slugifies to the empty string, so testing the name for
    truthiness picks it and then falls off the end.

    Suffixed rather than randomised on collision, so the second "Acme" is
    `acme-2` - still something a human recognises.
    """
    base = next((s for s in (_slugify(c) for c in candidates) if s), "") or "org"
    # create_org's validator wants at least three characters; a two-letter name
    # is a real one ("Jo"), so pad it rather than discard it.
    if len(base) < 3:
        base = f"{base}-org"

    slug, n = base, 1
    while session.query(Organization).filter(Organization.slug == slug).one_or_none():
        n += 1
        slug = f"{base}-{n}"
    return slug


def provision(session, user: User, org_name: str | None = None) -> Organization:
    """Give a brand-new account a tenant of its own, with that account as owner.

    The path every "signed up just now" flow ends at. A system principal because
    there is no existing tenant for the caller to be scoped to yet - which is the
    whole situation this function exists for.

    Naming is decided here rather than by the caller because everything needed is
    already on `user`: the display name if it gives a usable slug, the local part
    of the address otherwise.
    """
    local = user.email.split("@")[0]
    org = create_org(session, Principal.system(),
                     name=(org_name or "").strip() or local,
                     slug=_unique_slug(session, org_name, local))
    auth_core.add_membership(session, user, org, "owner")
    return org


def orgs_for_user(session, user_id: int) -> list[Organization]:
    """Every organisation a user belongs to - what the org switcher renders.

    Takes a user_id rather than a Principal because it answers a question about
    an account rather than about a tenant: the caller is asking which tenants
    they may enter, so it cannot be scoped to the one they are in.
    """
    return (session.query(Organization)
            .join(Membership, Membership.org_id == Organization.id)
            .filter(Membership.user_id == user_id)
            .order_by(Organization.name)
            .all())