docs: verify services

[Harshal6927]

• Verify service docs

[cofin]

You'll need to use Optional since we still need to show 3.9 support here.

[cofin]

Suggested change

	data = data_dict
	post = await self.to_model(data, "create")
	post = await self.to_model(data_dict, "create")

[cofin]

We should probably make the override mirror the service spec properly so there's no linting issues on the copy/paste

[cofin]

This needs to be remove or modified to mirror exactly how the TeamService works in fullstack. Here's the service for reference:

class TeamService(service.SQLAlchemyAsyncRepositoryService[m.Team]):
    """Team Service."""

    class Repo(repository.SQLAlchemyAsyncSlugRepository[m.Team]):
        """Team Repository."""

        model_type = m.Team

    repository_type = Repo
    match_fields = ["name"]

    async def to_model_on_create(self, data: ModelDictT[m.Team]) -> ModelDictT[m.Team]:
        data = service.schema_dump(data)
        data = await self._populate_slug(data)
        return await self._populate_with_owner_and_tags(data, "create")

    async def to_model_on_update(self, data: ModelDictT[m.Team]) -> ModelDictT[m.Team]:
        data = service.schema_dump(data)
        data = await self._populate_slug(data)
        return await self._populate_with_owner_and_tags(data, "update")

    async def to_model_on_upsert(self, data: ModelDictT[m.Team]) -> ModelDictT[m.Team]:
        data = service.schema_dump(data)
        data = await self._populate_slug(data)
        return await self._populate_with_owner_and_tags(data, "upsert")

    @staticmethod
    def can_view_all(user: m.User) -> bool:
        return bool(
            user.is_superuser
            or any(
                assigned_role.role.name
                for assigned_role in user.roles
                if assigned_role.role.name == constants.SUPERUSER_ACCESS_ROLE
            ),
        )

    async def _populate_slug(self, data: ModelDictT[m.Team]) -> ModelDictT[m.Team]:
        if service.is_dict_without_field(data, "slug") and service.is_dict_with_field(data, "name"):
            data["slug"] = await self.repository.get_available_slug(data["name"])
        return data

    async def _populate_with_owner_and_tags(
        self,
        data: ModelDictT[m.Team],
        operation: str | None,
    ) -> ModelDictT[m.Team]:
        if service.is_dict(data):
            owner_id: UUID | None = data.pop("owner_id", None)
            owner: m.User | None = data.pop("owner", None)
            input_tags: list[str] = data.pop("tags", [])
            data["id"] = data.get("id", uuid4())
            data = await super().to_model(data)
            if input_tags:
                existing_tags = [tag.name for tag in data.tags]
                tags_to_remove = [tag for tag in data.tags if tag.name not in input_tags]
                tags_to_add = [tag for tag in input_tags if tag not in existing_tags]
                for tag_rm in tags_to_remove:
                    data.tags.remove(tag_rm)
                data.tags.extend(
                    [
                        await m.Tag.as_unique_async(self.repository.session, name=tag_text, slug=slugify(tag_text))
                        for tag_text in tags_to_add
                    ],
                )
            if owner or owner_id:
                for member in data.members:
                    if member.user_id == owner.id if owner is not None else owner_id and not member.is_owner:
                        member.is_owner = True
                        member.role = m.TeamRoles.ADMIN
                        break
                else:
                    owner_data: dict[str, Any] = {"user": owner} if owner else {"user_id": owner_id}
                    data.members.append(m.TeamMember(**owner_data, role=m.TeamRoles.ADMIN, is_owner=True))
        return data

[cofin]

What do we want to demonstrate here exactly? this looks like all the default behavior you can do out of the box.

We should consider removing it to reduce confusion if there's nothing new.

[cofin]

Did we remove the slug example? this seems beneficial. Here's the TagsService for reference:

class TagService(service.SQLAlchemyAsyncRepositoryService[m.Tag]):
    """Handles basic lookup operations for an Tag."""

    class Repo(repository.SQLAlchemyAsyncSlugRepository[m.Tag]):
        """Tag Repository."""

        model_type = m.Tag

    repository_type = Repo
    match_fields = ["name"]

    async def to_model_on_create(self, data: service.ModelDictT[m.Tag]) -> service.ModelDictT[m.Tag]:
        data = service.schema_dump(data)
        if service.is_dict_without_field(data, "slug"):
            data["slug"] = await self.repository.get_available_slug(data["name"])
        return data

    async def to_model_on_update(self, data: service.ModelDictT[m.Tag]) -> service.ModelDictT[m.Tag]:
        data = service.schema_dump(data)
        if service.is_dict_without_field(data, "slug") and service.is_dict_with_field(data, "name"):
            data["slug"] = await self.repository.get_available_slug(data["name"])
        return data

    async def to_model_on_upsert(self, data: service.ModelDictT[m.Tag]) -> service.ModelDictT[m.Tag]:
        data = service.schema_dump(data)
        if service.is_dict_without_field(data, "slug") and (tag_name := data.get("name")) is not None:
            data["slug"] = await self.repository.get_available_slug(tag_name)
        return data

[cofin]

I'm not sure 2 level of arrays is the best thing to demo here.

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud