Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
114 changes: 90 additions & 24 deletions services/apps/git_integration/src/crowdgit/database/crud.py
Original file line number Diff line number Diff line change
Expand Up @@ -720,6 +720,20 @@ async def find_many_organization_ids_by_identities(identities: list[dict]) -> li
return results


async def fetch_organizations(org_ids: list[str]) -> list[dict]:
if not org_ids:
return []

return await query(
"""
SELECT id, "isAffiliationBlocked"
FROM organizations
WHERE id = ANY($1::uuid[])
""",
(org_ids,),
)


async def fetch_member_organizations(member_ids: list[str]) -> list[dict]:
if not member_ids:
return []
Expand Down Expand Up @@ -751,9 +765,9 @@ async def fetch_segment_affiliations(member_ids: list[str], segment_id: str) ->
)


async def insert_member_organizations(rows: list[dict]) -> None:
async def insert_member_organizations(rows: list[dict]) -> list[dict]:
if not rows:
return
return []

undated_rows: list[tuple] = []
open_ended_rows: list[tuple] = []
Expand All @@ -767,8 +781,10 @@ async def insert_member_organizations(rows: list[dict]) -> None:
row.get("date_end"),
row["source"],
)

date_start = row.get("date_start")
date_end = row.get("date_end")

if date_start is None and date_end is None:
undated_rows.append(params)
elif date_end is None:
Expand All @@ -787,41 +803,91 @@ async def insert_member_organizations(rows: list[dict]) -> None:
"createdAt",
"updatedAt"
)
VALUES ($1, $2, $3, $4, NULL, $5, NOW(), NOW())
"""

if undated_rows:
sql = (
insert_sql
+ """
returning_sql = """
RETURNING id, "memberId", "organizationId"
"""

buckets = [
(
undated_rows,
"""
ON CONFLICT ("memberId", "organizationId")
WHERE ("dateStart" IS NULL AND "dateEnd" IS NULL AND "deletedAt" IS NULL)
DO NOTHING
"""
)
await executemany(sql, undated_rows)

if open_ended_rows:
sql = (
insert_sql
+ """
""",
),
(
open_ended_rows,
"""
ON CONFLICT ("memberId", "organizationId", "dateStart")
WHERE ("dateEnd" IS NULL AND "deletedAt" IS NULL)
DO NOTHING
"""
)
await executemany(sql, open_ended_rows)

if dated_rows:
sql = (
insert_sql
+ """
""",
),
(
dated_rows,
"""
ON CONFLICT ("memberId", "organizationId", "dateStart", "dateEnd")
WHERE ("deletedAt" IS NULL)
DO NOTHING
""",
),
]

created_rows: list[dict] = []

for bucket_rows, conflict_sql in buckets:
if not bucket_rows:
continue

values_parts: list[str] = []
params: list = []
param_index = 1

for member_id, organization_id, date_start, date_end, source in bucket_rows:
values_parts.append(
f"(${param_index}, ${param_index + 1}, ${param_index + 2}, "
f"${param_index + 3}, NULL, ${param_index + 4}, NOW(), NOW())"
)
params.extend([member_id, organization_id, date_start, date_end, source])
param_index += 5

created_rows.extend(
await query(
insert_sql + f" VALUES {', '.join(values_parts)}" + conflict_sql + returning_sql,
tuple(params),
)
)

return created_rows


async def insert_member_organization_affiliation_overrides(rows: list[dict]) -> None:
if not rows:
return

await executemany(
"""
INSERT INTO "memberOrganizationAffiliationOverrides"(
id,
"memberId",
"memberOrganizationId",
"allowAffiliation"
)
await executemany(sql, dated_rows)
VALUES (gen_random_uuid(), $1, $2, $3)
ON CONFLICT ("memberId", "memberOrganizationId") DO NOTHING
""",
[
(
row["member_id"],
row["member_organization_id"],
row["allow_affiliation"],
)
for row in rows
],
)


async def insert_member_segment_affiliations(rows: list[dict]) -> None:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -11,10 +11,12 @@

from crowdgit.database.crud import (
fetch_member_organizations,
fetch_organizations,
fetch_segment_affiliations,
find_many_member_ids_by_identities,
find_many_organization_ids_by_identities,
get_repo_affiliation_registry,
insert_member_organization_affiliation_overrides,
insert_member_organizations,
insert_member_segment_affiliations,
save_service_execution,
Expand Down Expand Up @@ -870,6 +872,16 @@ async def apply_affiliations(
else:
segment_affiliations_by_member.setdefault(member_id, []).append(row)

blocked_org_ids: set[str] = set()
if resolved_stints:
org_ids = list({organization_id for _, organization_id, _ in resolved_stints})
organizations = await fetch_organizations(org_ids)
blocked_org_ids = {
str(organization["id"])
for organization in organizations
if organization["isAffiliationBlocked"]
}

mo_inserts: list[dict] = []
msa_inserts: list[dict] = []

Expand All @@ -896,10 +908,14 @@ async def apply_affiliations(
}
)

if not self.has_existing_stint(
existing_msas, organization_id, date_start, date_end
) and not self.is_blocked_by_deleted_row(
deleted_msas, organization_id, date_start, date_end
if (
str(organization_id) not in blocked_org_ids
and not self.has_existing_stint(
existing_msas, organization_id, date_start, date_end
)
and not self.is_blocked_by_deleted_row(
deleted_msas, organization_id, date_start, date_end
)
):
msa_inserts.append(
{
Expand All @@ -911,7 +927,21 @@ async def apply_affiliations(
}
)

await insert_member_organizations(mo_inserts)
created_member_organizations = await insert_member_organizations(mo_inserts)

if blocked_org_ids:
override_rows = [
{
"member_id": mo["memberId"],
"member_organization_id": mo["id"],
"allow_affiliation": False,
}
for mo in created_member_organizations
if str(mo["organizationId"]) in blocked_org_ids
]
if override_rows:
await insert_member_organization_affiliation_overrides(override_rows)

await insert_member_segment_affiliations(msa_inserts)

async def process_affiliations(
Expand Down
Loading