LITE-29260: Adding schemas for Credentials, plus generic schemas.

connect_bi_reporter/credentials/api/schemas.py

@@ -0,0 +1,16 @@
+from connect_bi_reporter.schemas import Events, NonNullSchema, ReferenceSchema
+
+
+class CredentialCreateSchema(NonNullSchema):
+    name: str
+    connection_string: str
+
+
+class CredentialListSchema(ReferenceSchema):
+    name: str
+    owner: ReferenceSchema
+    events: Events
+
+
+class CredentialGetSchema(CredentialListSchema):
+    connection_string: str

connect_bi_reporter/schemas.py

@@ -1,5 +1,45 @@
-# -*- coding: utf-8 -*-
-#
-# Copyright (c) 2024, Ingram Micro
-# All rights reserved.
-#
+from datetime import datetime
+from typing import Dict, Optional, Union
+
+from pydantic import BaseModel, root_validator
+
+_By = Optional[Union[str, Dict[str, str]]]
+Events = Dict[str, Dict[str, Union[datetime, _By]]]
+
+
+def clean_empties_from_dict(data):
+    """
+    Removes inplace all the fields that are None or empty dicts in data.
+    Returns param data, that was modified inplace.
+    If the param is not a dict, will return the param unmodified.
+    :param data: dict
+    :rtype: dict
+    """
+    if not isinstance(data, dict):
+        return data
+
+    for key in list(data.keys()):
+        value = data[key]
+        if isinstance(value, dict):
+            clean_empties_from_dict(value)
+            value = data[key]
+        if not value:
+            del data[key]
+    return data
+
+
+class NonNullSchema(BaseModel):
+    def dict(self, *args, **kwargs):
+        kwargs['exclude_none'] = True
+        return super().dict(*args, **kwargs)
+
+    @root_validator(pre=True)
+    def validate_events(cls, values):
+        events = values.get('events')
+        if events:
+            values['events'] = clean_empties_from_dict(events)
+        return values
+
+
+class ReferenceSchema(NonNullSchema):
+    id: str

tests/credentials/api/test_schemas.py

@@ -0,0 +1,61 @@
+from connect_bi_reporter.credentials.api.schemas import (
+    CredentialCreateSchema, CredentialGetSchema, CredentialListSchema,
+)
+
+
+def test_credentials_create_schema():
+    serializer = CredentialCreateSchema(
+        name='Creating credentials',
+        connection_string="D=https;AccountName=AN;AccountKey=AK;EndpointSuffix=core.windows.net",
+    )
+    assert serializer.dict() == {
+        'name': 'Creating credentials',
+        'connection_string': "D=https;AccountName=AN;AccountKey=AK;EndpointSuffix=core.windows.net",
+    }
+
+
+def test_credentials_list_schema(credential_factory):
+    credential = credential_factory()
+    serializer = CredentialListSchema(
+        id=credential.id,
+        name=credential.name,
+        owner={'id': credential.account_id},
+        events={
+            'created': {'at': credential.created_at, 'by': {'id': credential.created_by}},
+            'updated': {'at': credential.updated_at},
+        },
+    )
+    assert serializer.dict() == {
+        'id': credential.id,
+        'name': credential.name,
+        'owner': {'id': credential.account_id},
+        'events': {
+            'created': {'at': credential.created_at, 'by': {'id': credential.created_by}},
+            'updated': {'at': credential.updated_at},
+        },
+    }
+
+
+def test_credentials_get_schema(credential_factory):
+    credential = credential_factory()
+    serializer = CredentialGetSchema(
+        id=credential.id,
+        name=credential.name,
+        connection_string=credential.connection_string,
+        owner={'id': credential.account_id},
+        events={
+            'created': {'at': credential.created_at, 'by': {'id': credential.created_by}},
+            'updated': {'at': credential.updated_at, 'by': {'id': credential.updated_by}},
+        },
+    )
+
+    assert serializer.dict() == {
+        'id': credential.id,
+        'name': credential.name,
+        'connection_string': credential.connection_string,
+        'owner': {'id': credential.account_id},
+        'events': {
+            'created': {'at': credential.created_at, 'by': {'id': credential.created_by}},
+            'updated': {'at': credential.updated_at, 'by': {'id': credential.updated_by}},
+        },
+    }