34
35
from mailman.config import config
35
36
from mailman.core.constants import system_preferences
36
37
from mailman.interfaces.address import InvalidEmailAddressError
37
from mailman.interfaces.listmanager import IListManager, NoSuchListError
38
from mailman.interfaces.listmanager import (
39
IListManager, ListDeletedEvent, NoSuchListError)
38
40
from mailman.interfaces.member import DeliveryMode
39
41
from mailman.interfaces.subscriptions import (
40
42
ISubscriptionService, MissingUserError)
93
95
assert members.count() == 1, 'Too many matching members'
96
def find_members(self, subscriber, fqdn_listname=None, role=None):
98
def find_members(self, subscriber=None, fqdn_listname=None, role=None):
97
99
"""See `ISubscriptionService`."""
98
100
# If `subscriber` is a user id, then we'll search for all addresses
99
101
# which are controlled by the user, otherwise we'll just search for
100
102
# the given address.
101
103
user_manager = getUtility(IUserManager)
103
if isinstance(subscriber, basestring):
104
# subscriber is an email address.
105
address = user_manager.get_address(subscriber)
106
user = user_manager.get_user(subscriber)
107
# This probably could be made more efficient.
108
if address is None or user is None:
110
or_clause = Or(Member.address_id == address.id,
111
Member.user_id == user.id)
113
# subscriber is a user id.
114
user = user_manager.get_user_by_id(unicode(subscriber))
115
address_ids = list(address.id for address in user.addresses
116
if address.id is not None)
117
if len(address_ids) == 0 or user is None:
119
or_clause = Or(Member.user_id == user.id,
120
Member.address_id.is_in(address_ids))
121
# The rest is the same, based on the given criteria.
122
if fqdn_listname is None and role is None:
124
elif fqdn_listname is None:
125
query = And(Member.role == role, or_clause)
127
query = And(Member.mailing_list == fqdn_listname, or_clause)
129
query = And(Member.mailing_list == fqdn_listname,
132
results = config.db.store.find(Member, query)
104
if subscriber is None and fqdn_listname is None and role is None:
106
# Querying for the subscriber is the most complicated part, because
107
# the parameter can either be an email address or a user id.
109
if subscriber is not None:
110
if isinstance(subscriber, basestring):
111
# subscriber is an email address.
112
address = user_manager.get_address(subscriber)
113
user = user_manager.get_user(subscriber)
114
# This probably could be made more efficient.
115
if address is None or user is None:
117
query.append(Or(Member.address_id == address.id,
118
Member.user_id == user.id))
120
# subscriber is a user id.
121
user = user_manager.get_user_by_id(unicode(subscriber))
122
address_ids = list(address.id for address in user.addresses
123
if address.id is not None)
124
if len(address_ids) == 0 or user is None:
126
query.append(Or(Member.user_id == user.id,
127
Member.address_id.is_in(address_ids)))
128
# Calculate the rest of the query expression, which will get And'd
129
# with the Or clause above (if there is one).
130
if fqdn_listname is not None:
131
query.append(Member.mailing_list == fqdn_listname)
133
query.append(Member.role == role)
134
results = config.db.store.find(Member, And(*query))
133
135
return sorted(results, key=_membership_sort_key)
135
137
def __iter__(self):
177
179
raise NoSuchListError(fqdn_listname)
178
180
# XXX for now, no notification or user acknowledgement.
179
181
delete_member(mlist, address, False, False)
185
def handle_ListDeletedEvent(event):
186
"""Delete a mailing list's members when the list is deleted."""
188
if not isinstance(event, ListDeletedEvent):
190
# Find all the members still associated with the mailing list.
191
members = getUtility(ISubscriptionService).find_members(
192
fqdn_listname=event.fqdn_listname)
193
for member in members:
added
removed
src/mailman/app/subscriptions.py
