Change API to be able to assign roles to namespaces

conda-store-server/conda_store_server/alembic/versions/771180018e1b_add_v2_role_mappings.py

@@ -0,0 +1,39 @@
+"""add v2 role mappings
+
+Revision ID: 771180018e1b
+Revises: 30b37e725c32
+Create Date: 2023-11-29 09:02:35.835664
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "771180018e1b"
+down_revision = "30b37e725c32"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "namespace_role_mapping_v2",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("namespace_id", sa.Integer(), nullable=False),
+        sa.Column("other_namespace_id", sa.Integer(), nullable=False),
+        sa.Column("role", sa.Unicode(length=255), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["namespace_id"],
+            ["namespace.id"],
+        ),
+        sa.ForeignKeyConstraint(
+            ["other_namespace_id"],
+            ["namespace.id"],
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("namespace_id", "other_namespace_id", name="_uc"),
+    )
+
+
+def downgrade():
+    op.drop_table("namespace_role_mapping_v2")

conda-store-server/conda_store_server/api.py

@@ -3,6 +3,7 @@
 
 from conda_store_server import conda_utils, orm, schema, utils
 from sqlalchemy import distinct, func, null, or_
+from sqlalchemy.orm import aliased
 
 
 def list_namespaces(db, show_soft_deleted: bool = False):
@@ -82,6 +83,176 @@ def update_namespace(
     return namespace
 
 
+# v2 API
+def update_namespace_metadata(
+    db,
+    name: str,
+    metadata_: Dict[str, Any] = None,
+):
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    if metadata_ is not None:
+        namespace.metadata_ = metadata_
+
+    return namespace
+
+
+# v2 API
+def get_namespace_roles(
+    db,
+    name: str,
+):
+    """Which namespaces can access namespace 'name'?"""
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    nrm = aliased(orm.NamespaceRoleMappingV2)
+    this = aliased(orm.Namespace)
+    other = aliased(orm.Namespace)
+    q = (
+        db.query(nrm.id, this.name, other.name, nrm.role)
+        .filter(nrm.namespace_id == namespace.id)
+        .filter(nrm.namespace_id == this.id)
+        .filter(nrm.other_namespace_id == other.id)
+        .all()
+    )
+    return [schema.NamespaceRoleMappingV2.from_list(x) for x in q]
+
+
+# v2 API
+def get_other_namespace_roles(
+    db,
+    name: str,
+):
+    """To which namespaces does namespace 'name' have access?"""
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    nrm = aliased(orm.NamespaceRoleMappingV2)
+    this = aliased(orm.Namespace)
+    other = aliased(orm.Namespace)
+    q = (
+        db.query(nrm.id, this.name, other.name, nrm.role)
+        .filter(nrm.other_namespace_id == namespace.id)
+        .filter(nrm.namespace_id == this.id)
+        .filter(nrm.other_namespace_id == other.id)
+        .all()
+    )
+    return [schema.NamespaceRoleMappingV2.from_list(x) for x in q]
+
+
+# v2 API
+def delete_namespace_roles(
+    db,
+    name: str,
+):
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    nrm = orm.NamespaceRoleMappingV2
+    db.query(nrm).filter(nrm.namespace_id == namespace.id).delete()
+
+
+# v2 API
+def get_namespace_role(
+    db,
+    name: str,
+    other: str,
+):
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    other_namespace = get_namespace(db, other)
+    if other_namespace is None:
+        raise ValueError(f"Namespace='{other}' not found")
+
+    nrm = aliased(orm.NamespaceRoleMappingV2)
+    this = aliased(orm.Namespace)
+    other = aliased(orm.Namespace)
+    q = (
+        db.query(nrm.id, this.name, other.name, nrm.role)
+        .filter(nrm.namespace_id == namespace.id)
+        .filter(nrm.other_namespace_id == other_namespace.id)
+        .filter(nrm.namespace_id == this.id)
+        .filter(nrm.other_namespace_id == other.id)
+        .first()
+    )
+    if q is None:
+        return None
+    return schema.NamespaceRoleMappingV2.from_list(q)
+
+
+# v2 API
+def create_namespace_role(
+    db,
+    name: str,
+    other: str,
+    role: str,
+):
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    other_namespace = get_namespace(db, other)
+    if other_namespace is None:
+        raise ValueError(f"Namespace='{other}' not found")
+
+    db.add(
+        orm.NamespaceRoleMappingV2(
+            namespace_id=namespace.id,
+            other_namespace_id=other_namespace.id,
+            role=role,
+        )
+    )
+
+
+# v2 API
+def update_namespace_role(
+    db,
+    name: str,
+    other: str,
+    role: str,
+):
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    other_namespace = get_namespace(db, other)
+    if other_namespace is None:
+        raise ValueError(f"Namespace='{other}' not found")
+
+    nrm = orm.NamespaceRoleMappingV2
+    db.query(nrm).filter(nrm.namespace_id == namespace.id).filter(
+        nrm.other_namespace_id == other_namespace.id
+    ).update({"role": role})
+
+
+# v2 API
+def delete_namespace_role(
+    db,
+    name: str,
+    other: str,
+):
+    namespace = get_namespace(db, name)
+    if namespace is None:
+        raise ValueError(f"Namespace='{name}' not found")
+
+    other_namespace = get_namespace(db, other)
+    if other_namespace is None:
+        raise ValueError(f"Namespace='{other}' not found")
+
+    nrm = orm.NamespaceRoleMappingV2
+    db.query(nrm).filter(nrm.namespace_id == namespace.id).filter(
+        nrm.other_namespace_id == other_namespace.id
+    ).delete()
+
+
 def delete_namespace(db, name: str = None, id: int = None):
     namespace = get_namespace(db, name=name, id=id)
     if namespace:

conda-store-server/conda_store_server/orm.py

@@ -89,6 +89,37 @@ def validate_role(self, key, role):
         return role
 
 
+class NamespaceRoleMappingV2(Base):
+    """Mapping between roles and namespaces"""
+
+    __tablename__ = "namespace_role_mapping_v2"
+
+    id = Column(Integer, primary_key=True)
+    # Provides access to this namespace
+    namespace_id = Column(Integer, ForeignKey("namespace.id"), nullable=False)
+    namespace = relationship(Namespace, foreign_keys=[namespace_id])
+
+    # ... for other namespace
+    other_namespace_id = Column(Integer, ForeignKey("namespace.id"), nullable=False)
+    other_namespace = relationship(Namespace, foreign_keys=[other_namespace_id])
+
+    # ... with this role, like 'viewer'
+    role = Column(Unicode(255), nullable=False)
+
+    @validates("role")
+    def validate_role(self, key, role):
+        if role not in ["admin", "viewer", "developer"]:
+            raise ValueError(f"invalid role={role}")
+        return role
+
+    __table_args__ = (
+        # Ensures no duplicates can be added with this combination of fields.
+        # Note: this doesn't add role because role needs to be unique for each
+        # pair of ids.
+        UniqueConstraint("namespace_id", "other_namespace_id", name="_uc"),
+    )
+
+
 class Specification(Base):
     """The specifiction for a given conda environment"""
 

conda-store-server/conda_store_server/schema.py

@@ -38,8 +38,9 @@ class Permissions(enum.Enum):
     NAMESPACE_READ = "namespace::read"
     NAMESPACE_UPDATE = "namespace::update"
     NAMESPACE_DELETE = "namespace::delete"
-    NAMESPACE_ROLE_MAPPING_READ = "namespace-role-mapping::read"
     NAMESPACE_ROLE_MAPPING_CREATE = "namespace-role-mapping::create"
+    NAMESPACE_ROLE_MAPPING_READ = "namespace-role-mapping::read"
+    NAMESPACE_ROLE_MAPPING_UPDATE = "namespace-role-mapping::update"
     NAMESPACE_ROLE_MAPPING_DELETE = "namespace-role-mapping::delete"
     SETTING_READ = "setting::read"
     SETTING_UPDATE = "setting::update"
@@ -102,6 +103,20 @@ class Config:
         orm_mode = True
 
 
+class NamespaceRoleMappingV2(BaseModel):
+    id: int
+    namespace: str
+    other_namespace: str
+    role: str
+
+    class Config:
+        orm_mode = True
+
+    @classmethod
+    def from_list(cls, lst):
+        return cls(**{k: v for k, v in zip(cls.__fields__.keys(), lst)})
+
+
 class Namespace(BaseModel):
     id: int
     name: constr(regex=f"^[{ALLOWED_CHARACTERS}]+$")  # noqa: F722
@@ -616,6 +631,23 @@ class APIGetNamespace(APIResponse):
     data: Namespace
 
 
+# POST /api/v1/namespace/{name}/role
+class APIPostNamespaceRole(BaseModel):
+    other_namespace: str
+    role: str
+
+
+# PUT /api/v1/namespace/{name}/role
+class APIPutNamespaceRole(BaseModel):
+    other_namespace: str
+    role: str
+
+
+# DELETE /api/v1/namespace/{name}/role
+class APIDeleteNamespaceRole(BaseModel):
+    other_namespace: str
+
+
 # GET /api/v1/environment
 class APIListEnvironment(APIPaginatedResponse):
     data: List[Environment]

conda-store-server/conda_store_server/server/app.py

@@ -254,6 +254,14 @@ async def http_exception_handler(request, exc):
                 status_code=exc.status_code,
             )
 
+        # Prints exceptions to the terminal
+        # https://fastapi.tiangolo.com/tutorial/handling-errors/#re-use-fastapis-exception-handlers
+        # https://github.com/tiangolo/fastapi/issues/1241
+        @app.exception_handler(Exception)
+        async def exception_handler(request, exc):
+            print(exc)
+            return await http_exception_handler(request, exc)
+
         app.include_router(
             self.authentication.router,
             prefix=trim_slash(self.url_prefix),