def _check_pglogical_schema_usage_privilege()

in src/dma/collector/workflows/readiness_check/_postgres/main.py [0:0]

• 10 lines of code
• 3 McCabe index (conditional complexity)

    def _check_pglogical_schema_usage_privilege(self, db_name: str) -> str | None:
        result = self.local_db.sql(
            "select has_schema_usage_privilege from collection_postgres_pglogical_schema_usage_privilege where database_name = $db_name",
            params={"db_name": db_name},
        ).fetchone()
        if result is None:
            return "Empty result reading pglogical schema usage privilege for the user"
        if not result[0]:
            return "user doesn't have USAGE privilege on schema pglogical"
        return None