feat(sponsorships): team tags on sponsors, with a tag filter and desk sidebar

.gitignore

@@ -20,7 +20,7 @@ wheels/
 *.egg-info/
 .installed.cfg
 *.egg
-tags
+/tags
 MANIFEST
 
 # Environments

buzz/api/filters/conditions.py

@@ -16,6 +16,9 @@
 	"Attach": "file",
 	"Attach Image": "file",
 }
+# A record's Buzz Tags, matched through its Buzz Tag Link rows.
+TAGS_KEY = "tags"
+
 NEGATED_OPERATORS = {"not in": "in", "not like": "like"}
 # Answers saved through `str()` carry Python's spelling: "True" for a tick, ['a', 'b'] for a list.
 CHECKED_VALUES = ["1", "True", "true"]
@@ -64,6 +67,10 @@ def filter_field(key, label, fieldtype, options=(), section="standard") -> Filte
 	)
 
 
+def tags_field(tags) -> FilterField:
+	return filter_field(TAGS_KEY, _("Tags"), "Link", [(tag.name, tag.label) for tag in tags])
+
+
 def question_fields(questions) -> list[FilterField]:
 	"""Buzz Form Field rows as filters; answers are matched in Additional Field by fieldname."""
 	return [
@@ -125,12 +132,21 @@ def is_valid(self, triple) -> bool:
 	def translate(self, key: str, operator: str, value) -> list | None:
 		if is_blank(operator, value):
 			return None
+		if key == TAGS_KEY:
+			return ["name", operator, self.tagged_names(value)]
 		field = self.fields[key]
 		if field.section == "question":
 			doctype, link_field = self.answer_parents.get(key, (self.doctype, "name"))
 			return AnswerCondition(doctype, link_field, field, operator, value).name_filter()
 		return [key, operator, f"%{value}%" if "like" in operator else value]
 
+	def tagged_names(self, tags: list[str]) -> list[str]:
+		return frappe.get_all(
+			"Buzz Tag Link",
+			filters={"document_type": self.doctype, "tag": ["in", tags]},
+			pluck="document_name",
+		)
+
 
 class AnswerCondition:
 	"""One condition on a custom question, resolved to the parents whose answer matches.

buzz/api/sponsorships/__init__.py

@@ -13,10 +13,12 @@
 	EnquiryDetail,
 	EventEnquiriesResponse,
 	EventSponsorshipsResponse,
+	EventSponsorsResponse,
 	SponsorshipDetailsResponse,
 	SponsorshipListItem,
 )
 from buzz.api.sponsorships.services import SponsorshipService, list_user_enquiries
+from buzz.api.sponsorships.sponsors import event_sponsors
 
 
 # nosemgrep: guest-whitelisted-method
@@ -63,6 +65,14 @@ def get_event_sponsorships(event: str) -> EventSponsorshipsResponse:
 	return event_sponsorships(event)
 
 
+@frappe.whitelist(methods=["GET"])
+def get_event_sponsors(
+	event: str, search: str | None = None, filters: str | None = None, order: str = "desc"
+) -> EventSponsorsResponse:
+	"""`filters` is a JSON list of `[field, operator, value]`, the string the dashboard keeps in its URL."""
+	return event_sponsors(event, search, filters, order)
+
+
 @frappe.whitelist(methods=["GET"])
 def get_event_sponsorship_enquiries(
 	event: str,

buzz/api/sponsorships/manage.py

@@ -5,23 +5,13 @@
 from buzz.api.events.services import ensure_event_team_access
 from buzz.api.sponsorships.schemas import (
 	EnquiryFormState,
-	EventSponsorItem,
 	EventSponsorshipsResponse,
 	TierItem,
 )
+from buzz.api.sponsorships.sponsors import sponsor_items, sponsor_rows, sponsor_tags
 from buzz.permissions import has_team_access
 
 TIER_FIELDS = ["name", "title", "enabled", "perks", "slots", {"prices": ["currency", "price"]}]
-SPONSOR_FIELDS = [
-	"name",
-	"company_name",
-	"company_logo",
-	"website",
-	"country",
-	"contact_email",
-	"enquiry",
-	"tier",
-]
 
 
 def event_sponsorships(event: str) -> EventSponsorshipsResponse:
@@ -36,28 +26,18 @@ def event_sponsorships(event: str) -> EventSponsorshipsResponse:
 		order_by="creation asc",
 		ignore_permissions=True,
 	)
-	sponsors = frappe.get_all(
-		"Event Sponsor",
-		filters={"event": event},
-		fields=SPONSOR_FIELDS,
-		order_by="creation asc",
-		ignore_permissions=True,
-	)
-	tier_titles = {tier.name: tier.title for tier in tiers}
+	sponsors = sponsor_rows({"event": event})
+	tier_titles = {str(tier.name): tier.title for tier in tiers}
 	sponsor_counts = Counter(sponsor.tier for sponsor in sponsors)
 
 	return EventSponsorshipsResponse(
 		title=doc.title,
 		can_write=has_team_access(doc.team, "write", frappe.session.user),
 		form=form_state(doc),
 		tiers=[TierItem(**tier, sponsor_count=sponsor_counts[tier.name]) for tier in tiers],
-		sponsors=[
-			EventSponsorItem(
-				**sponsor,
-				tier_title=tier_titles.get(sponsor.tier, sponsor.tier) if sponsor.tier else "",
-			)
-			for sponsor in sponsors
-		],
+		sponsors=sponsor_items(sponsors, tier_titles),
+		team=doc.team,
+		tags=sponsor_tags(event),
 	)
 
 

buzz/api/sponsorships/schemas.py

@@ -2,6 +2,7 @@
 
 from buzz.api.filters.schemas import FilterField
 from buzz.api.schemas import APIResponse
+from buzz.api.tags.schemas import TagItem
 
 
 class EnquiryFormState(APIResponse):
@@ -35,6 +36,7 @@ class EventSponsorItem(APIResponse):
 	enquiry: str | None
 	tier: str | None
 	tier_title: str
+	tags: list[TagItem]
 
 
 class EventEnquiryItem(APIResponse):
@@ -56,6 +58,17 @@ class EventSponsorshipsResponse(APIResponse):
 	form: EnquiryFormState | None
 	tiers: list[TierItem]
 	sponsors: list[EventSponsorItem]
+	team: str
+	# The team's sponsor tags, which the sponsor drawer offers.
+	tags: list[TagItem]
+
+
+class EventSponsorsResponse(APIResponse):
+	"""An event's sponsors after search and filters, with what the list can be filtered by."""
+
+	total: int
+	sponsors: list[EventSponsorItem]
+	filter_fields: list[FilterField]
 
 
 class EventEnquiriesResponse(APIResponse):

buzz/api/sponsorships/sponsors.py

@@ -0,0 +1,78 @@
+import frappe
+from frappe import _
+
+from buzz.api.events.services import ensure_event_team_access
+from buzz.api.filters.conditions import ListConditions, filter_field, tags_field
+from buzz.api.filters.schemas import FilterField
+from buzz.api.sponsorships.enquiries import search_filters
+from buzz.api.sponsorships.schemas import EventSponsorItem, EventSponsorsResponse
+from buzz.api.tags.schemas import TagItem
+from buzz.api.tags.services import tags_by_document, team_tags
+
+SPONSOR_FIELDS = [
+	"name",
+	"company_name",
+	"company_logo",
+	"website",
+	"country",
+	"contact_email",
+	"enquiry",
+	"tier",
+]
+
+
+def event_sponsors(
+	event: str, search: str | None = None, filters: str | None = None, order: str = "desc"
+) -> EventSponsorsResponse:
+	"""An event's sponsors narrowed by search and filters; few enough per event to skip paging."""
+	ensure_event_team_access(event)
+	filter_fields = sponsor_filter_fields(event)
+	conditions = ListConditions("Event Sponsor", filter_fields).frappe_filters(filters)
+	# Interpolated into ORDER BY, so it can only ever be one of two literals.
+	direction = "asc" if str(order).lower() == "asc" else "desc"
+	rows = sponsor_rows([["event", "=", event], *conditions], search_filters(search), direction)
+	return EventSponsorsResponse(
+		total=frappe.db.count("Event Sponsor", {"event": event}),
+		sponsors=sponsor_items(rows, tier_titles(event)),
+		filter_fields=filter_fields,
+	)
+
+
+def sponsor_rows(filters: list, or_filters: list | None = None, direction: str = "asc") -> list:
+	return frappe.get_all(
+		"Event Sponsor",
+		filters=filters,
+		or_filters=or_filters,
+		fields=SPONSOR_FIELDS,
+		order_by=f"creation {direction}, name {direction}",
+		ignore_permissions=True,
+	)
+
+
+def sponsor_items(rows: list, tier_titles: dict) -> list[EventSponsorItem]:
+	tags = tags_by_document("Event Sponsor", [row.name for row in rows])
+	return [
+		EventSponsorItem(
+			**row,
+			tags=tags.get(row.name, []),
+			tier_title=tier_titles.get(str(row.tier), row.tier) if row.tier else "",
+		)
+		for row in rows
+	]
+
+
+def tier_titles(event: str) -> dict[str, str]:
+	tiers = frappe.get_all("Sponsorship Tier", {"event": event}, ["name", "title"], order_by="creation")
+	return {str(tier.name): tier.title for tier in tiers}
+
+
+def sponsor_filter_fields(event: str) -> list[FilterField]:
+	return [
+		filter_field("tier", _("Tier"), "Link", list(tier_titles(event).items())),
+		tags_field(sponsor_tags(event)),
+	]
+
+
+def sponsor_tags(event: str) -> list[TagItem]:
+	"""Sponsor tags are the team's, shared by all its events."""
+	return team_tags(frappe.db.get_value("Buzz Event", event, "team"), "Event Sponsor")

buzz/api/sponsorships/test_manage.py

@@ -22,6 +22,7 @@
 	"enquiry",
 	"tier",
 	"tier_title",
+	"tags",
 }
 ENQUIRY_FIELDS = {
 	"name",

buzz/api/sponsorships/test_sponsors.py

@@ -0,0 +1,56 @@
+import json
+
+import frappe
+
+from buzz.api.events.exceptions import CannotManageEvent
+from buzz.api.sponsorships import get_event_sponsors, get_event_sponsorships
+from buzz.api.tags import set_tags
+from buzz.tests.base_test_cases import SponsorshipTestCase
+from buzz.tests.factories import BuzzEventFactory, BuzzTagFactory, EventSponsorFactory
+
+
+class TestSponsorList(SponsorshipTestCase):
+	@classmethod
+	def setUpClass(cls):
+		super().setUpClass()
+		cls.team = frappe.db.get_value("Buzz Event", cls.event, "team")
+
+	def setUp(self):
+		# Rollback is per class, so each test lists the sponsors of its own event.
+		self.event = str(BuzzEventFactory.create(team=self.team).name)
+		super().setUp()
+		self.acme = self.make_sponsor().name
+		self.globex = EventSponsorFactory.create(event=self.event, company_name="Globex").name
+		self.vip = BuzzTagFactory.create(team=self.team)
+		set_tags("Event Sponsor", self.acme, [self.vip.name])
+
+	def sponsors(self, *triples, search=None) -> set[str]:
+		response = get_event_sponsors(self.event, search=search, filters=json.dumps(triples))
+		return {sponsor.name for sponsor in response.sponsors}
+
+	def test_sponsor_lists_its_tags(self):
+		response = get_event_sponsorships(self.event)
+
+		sponsor = next(row for row in response.sponsors if row.name == self.acme)
+		self.assertEqual([tag.name for tag in sponsor.tags], [self.vip.name])
+		self.assertIn(self.vip.name, [tag.name for tag in response.tags])
+
+	def test_filter_by_tag(self):
+		self.assertEqual(self.sponsors(["tags", "in", [self.vip.name]]), {self.acme})
+		self.assertEqual(self.sponsors(["tags", "not in", [self.vip.name]]), {self.globex})
+
+	def test_filter_by_tier_and_search(self):
+		self.assertEqual(self.sponsors(["tier", "in", [str(self.tier.name)]]), {self.acme})
+		self.assertEqual(self.sponsors(search="glob"), {self.globex})
+
+	def test_tag_filter_offers_only_sponsor_tags_of_the_team(self):
+		event_tag = BuzzTagFactory.create(team=self.team, document_type="Buzz Event").name
+
+		fields = {field.key: field for field in get_event_sponsors(self.event).filter_fields}
+		offered = [option.value for option in fields["tags"].options]
+		self.assertIn(self.vip.name, offered)
+		self.assertNotIn(event_tag, offered)
+
+	def test_non_member_is_refused(self):
+		with self.set_user(self.make_stranger()), self.assertRaises(CannotManageEvent):
+			get_event_sponsors(self.event)

buzz/api/tags/__init__.py

@@ -0,0 +1,15 @@
+import frappe
+
+from buzz.api.tags.schemas import TagItem
+from buzz.api.tags.services import DocumentTags, create_team_tag
+
+
+@frappe.whitelist(methods=["POST"])
+def create_tag(team: str, document_type: str, label: str, color: str = "gray") -> TagItem:
+	return create_team_tag(team, document_type, label, color)
+
+
+@frappe.whitelist(methods=["POST"])
+def set_tags(document_type: str, document_name: str, tags: list[str]) -> list[TagItem]:
+	"""Replaces every tag on the record with `tags`, a list of Buzz Tag names."""
+	return DocumentTags(document_type, document_name).replace(tags)

buzz/api/tags/desk.py

@@ -0,0 +1,48 @@
+import frappe
+from frappe import _
+from frappe.desk.doctype.tag import tag as frappe_tags
+
+from buzz.api.tags.services import DocumentTags, create_team_tag, is_taggable
+from buzz.events.doctype.buzz_tag.buzz_tag import document_team
+
+# Desk's tag editor, routed to Buzz Tags on taggable records and to Frappe's tags everywhere else.
+
+
+@frappe.whitelist()
+def add_tag(tag: str, dt: str, dn: str, color: str | None = None):
+	if not is_taggable(dt):
+		return frappe_tags.add_tag(tag, dt, dn, color)
+	tags = DocumentTags(dt, dn)
+	added = create_team_tag(document_team(dt, dn), dt, tag)
+	tags.replace([*tags.names(), added.name])
+	return added.label
+
+
+@frappe.whitelist()
+def add_tags(tags: str | list[str], dt: str, docs: str | list[str], color: str | None = None):
+	if not is_taggable(dt):
+		return frappe_tags.add_tags(tags, dt, docs, color)
+	if not frappe.get_cached_value("User", frappe.session.user, "bulk_actions"):
+		frappe.throw(_("You are not allowed to perform bulk actions"), frappe.PermissionError)
+	for doc in frappe.parse_json(docs):
+		for tag in frappe.parse_json(tags):
+			add_tag(tag, dt, doc)
+
+
+@frappe.whitelist()
+def remove_tag(tag: str, dt: str, dn: str):
+	if not is_taggable(dt):
+		return frappe_tags.remove_tag(tag, dt, dn)
+	tags = DocumentTags(dt, dn)
+	removed = frappe.db.get_value(
+		"Buzz Tag", {"team": document_team(dt, dn), "document_type": dt, "label": tag}
+	)
+	tags.replace([name for name in tags.names() if name != removed])
+
+
+@frappe.whitelist()
+def get_tags(doctype: str, txt: str):
+	if not is_taggable(doctype):
+		return frappe_tags.get_tags(doctype, txt)
+	filters = {"document_type": doctype, "label": ["like", f"%{txt}%"]}
+	return sorted(set(frappe.get_list("Buzz Tag", filters=filters, pluck="label")))

buzz/api/tags/exceptions.py

@@ -0,0 +1,13 @@
+from frappe import _lt
+
+from buzz.api.exceptions import BuzzAPIError
+
+
+class NotTaggable(BuzzAPIError):
+	title = _lt("Cannot Tag")
+	message = _lt("These records cannot be tagged.")
+
+
+class EmptyTagLabel(BuzzAPIError):
+	title = _lt("Name Required")
+	message = _lt("Give the tag a name.")