feat: all unique constraints tested

src/iam/models.py

@@ -33,8 +33,13 @@ class Permission(Base):
 
 	service_id = Column(Integer, ForeignKey("service.id", ondelete="CASCADE"))
 
-	UniqueConstraint(
-		"service_id", "resource", "action", name="uniq_permission_resource_and_action"
+	__table_args__ = (
+		UniqueConstraint(
+			"service_id",
+			"resource",
+			"action",
+			name="uniq_permission_resource_and_action",
+		),
 	)
 
 	service_rel = relationship("Service", foreign_keys=[service_id])

src/iam/router.py

@@ -18,7 +18,6 @@ Endpoints:
 
 from fastapi import APIRouter, status
 from sqlalchemy.exc import IntegrityError
-from psycopg import errors
 
 from src.service.exceptions import ServiceNotFoundException
 from src.exceptions import ConflictException
@@ -264,12 +263,15 @@ async def create_new_permission(
 	if service_model is None:
 		raise ServiceNotFoundException(service_id=request_model.service_id)
 	perm_model = Perm(**request_model.__dict__)
+	db.add(perm_model)
 	try:
-		db.add(perm_model)
+		db.flush()
 	except IntegrityError as e:
-		if isinstance(e.orig, errors.UniqueViolation):
+		if (
+			getattr(e.orig, "pgcode", None) == "23505"  # Postgres unique violation
+			or "UNIQUE constraint failed" in str(e.orig)  # SQLite unique violation
+		):
 			raise ConflictException(message="Permission already exists")
-	db.flush()
 	response = {
 		"service_name": perm_model.service_name,
 		"resource": perm_model.resource,

src/organisation/router.py

@@ -20,7 +20,6 @@ from typing import Annotated
 
 from fastapi import APIRouter, status
 from fastapi.params import Query
-from psycopg.errors import UniqueViolation
 from sqlalchemy.exc import IntegrityError
 
 from src.contact.schemas import ContactModel
@@ -143,7 +142,10 @@ async def create_org(
 	try:
 		db.flush()
 	except IntegrityError as e:
-		if isinstance(e.orig, UniqueViolation):
+		if (
+			getattr(e.orig, "pgcode", None) == "23505"  # Postgres unique violation
+			or "UNIQUE constraint failed" in str(e.orig)  # SQLite unique violation
+		):
 			raise ConflictException(
 				message="Organisation with this name already exists"
 			)

src/service/router.py

@@ -9,7 +9,6 @@ Endpoints:
 """
 
 from fastapi import APIRouter, status
-from psycopg.errors import UniqueViolation
 from sqlalchemy.exc import IntegrityError
 
 from src.exceptions import ConflictException
@@ -87,7 +86,10 @@ async def register_service(
 	try:
 		db.flush()
 	except IntegrityError as e:
-		if isinstance(e.orig, UniqueViolation):
+		if (
+			getattr(e.orig, "pgcode", None) == "23505"  # Postgres unique violation
+			or "UNIQUE constraint failed" in str(e.orig)  # SQLite unique violation
+		):
 			raise ConflictException(message="Service with this name already exists")
 	response = ServiceWithKeySchema(**service_model.__dict__)
 	db.commit()