本文整理汇总了Python中temba.contacts.models.URN类的典型用法代码示例。如果您正苦于以下问题:Python URN类的具体用法?Python URN怎么用?Python URN使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了URN类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: create_contact
def create_contact(self, name=None, number=None, twitter=None, urn=None, is_test=False, **kwargs):
"""
Create a contact in the master test org
"""
urns = []
if number:
urns.append(URN.from_tel(number))
if twitter:
urns.append(URN.from_twitter(twitter))
if urn:
urns.append(urn)
if not name and not urns: # pragma: no cover
raise ValueError("Need a name or URN to create a contact")
kwargs['name'] = name
kwargs['urns'] = urns
kwargs['is_test'] = is_test
if 'org' not in kwargs:
kwargs['org'] = self.org
if 'user' not in kwargs:
kwargs['user'] = self.user
return Contact.get_or_create(**kwargs)
开发者ID:ianjuma,项目名称:rapidpro,代码行数:25,代码来源:tests.py
示例2: serialize_contact
def serialize_contact(contact):
from temba.contacts.models import URN
field_values = {}
for field in contact.org.cached_contact_fields.values():
field_values[field.key] = contact.get_field_json(field)
# augment URN values with preferred channel UUID as a parameter
urn_values = []
for u in contact.urns.order_by("-priority", "id"):
# for each URN we include the preferred channel as a query param if there is one
if u.channel and u.channel.is_active:
scheme, path, query, display = URN.to_parts(u.urn)
urn_str = URN.from_parts(scheme, path, query=urlencode({"channel": str(u.channel.uuid)}), display=display)
else:
urn_str = u.urn
urn_values.append(urn_str)
return {
"uuid": contact.uuid,
"id": contact.id,
"name": contact.name,
"language": contact.language,
"urns": urn_values,
"groups": [serialize_ref(group) for group in contact.user_groups.filter(is_active=True)],
"fields": field_values,
}
开发者ID:mxabierto,项目名称:rapidpro,代码行数:28,代码来源:serialize.py
示例3: create_contact
def create_contact(self, name=None, number=None, twitter=None, twitterid=None, urn=None, is_test=False, **kwargs):
"""
Create a contact in the master test org
"""
urns = []
if number:
urns.append(URN.from_tel(number))
if twitter:
urns.append(URN.from_twitter(twitter))
if twitterid:
urns.append(URN.from_twitterid(twitterid))
if urn:
urns.append(urn)
if not name and not urns: # pragma: no cover
raise ValueError("Need a name or URN to create a contact")
kwargs["name"] = name
kwargs["urns"] = urns
kwargs["is_test"] = is_test
if "org" not in kwargs:
kwargs["org"] = self.org
if "user" not in kwargs:
kwargs["user"] = self.user
return Contact.get_or_create_by_urns(**kwargs)
开发者ID:mxabierto,项目名称:rapidpro,代码行数:27,代码来源:base.py
示例4: normalize_urn
def normalize_urn(self, value):
if self.request.user.get_org().is_anon:
raise InvalidQueryError("URN lookups not allowed for anonymous organizations")
try:
return URN.identity(URN.normalize(value))
except ValueError:
raise InvalidQueryError("Invalid URN: %s" % value)
开发者ID:teehamaral,项目名称:rapidpro,代码行数:8,代码来源:views_base.py
示例5: test_resolve
def test_resolve(self, mock_lookup_user):
self.joe = self.create_contact("joe", twitter="therealjoe")
urn = self.joe.get_urns()[0]
# test no return value, should cause joe to be stopped
mock_lookup_user.return_value = []
resolve_twitter_ids()
self.joe.refresh_from_db()
urn.refresh_from_db()
self.assertTrue(self.joe.is_stopped)
self.assertIsNone(urn.display)
self.assertEqual("twitter:therealjoe", urn.identity)
self.assertEqual("therealjoe", urn.path)
self.joe.unstop(self.admin)
# test a real return value
mock_lookup_user.return_value = [dict(screen_name="TheRealJoe", id="123456")]
resolve_twitter_ids()
urn.refresh_from_db()
self.assertIsNone(urn.contact)
new_urn = self.joe.get_urns()[0]
self.assertEqual("twitterid:123456", new_urn.identity)
self.assertEqual("123456", new_urn.path)
self.assertEqual("therealjoe", new_urn.display)
self.assertEqual("twitterid:123456#therealjoe", new_urn.urn)
old_fred = self.create_contact("old fred", urn=URN.from_twitter("fred"))
new_fred = self.create_contact("new fred", urn=URN.from_twitterid("12345", screen_name="fred"))
mock_lookup_user.return_value = [dict(screen_name="fred", id="12345")]
resolve_twitter_ids()
# new fred shouldn't have any URNs anymore as he really is old_fred
self.assertEqual(0, len(new_fred.urns.all()))
# old fred should be unchanged
self.assertEqual("twitterid:12345", old_fred.urns.all()[0].identity)
self.jane = self.create_contact("jane", twitter="jane10")
mock_lookup_user.side_effect = Exception(
"Twitter API returned a 404 (Not Found), No user matches for specified terms."
)
resolve_twitter_ids()
self.jane.refresh_from_db()
self.assertTrue(self.jane.is_stopped)
self.sarah = self.create_contact("sarah", twitter="sarah20")
mock_lookup_user.side_effect = Exception("Unable to reach API")
resolve_twitter_ids()
self.sarah.refresh_from_db()
self.assertFalse(self.sarah.is_stopped)
开发者ID:mxabierto,项目名称:rapidpro,代码行数:58,代码来源:tests.py
示例6: to_internal_value
def to_internal_value(self, data):
try:
normalized = URN.normalize(data)
if not URN.validate(normalized):
raise ValueError()
except ValueError:
raise serializers.ValidationError("Invalid URN: %s. Ensure phone numbers contain country codes." % data)
return normalized
开发者ID:eHealthAfrica,项目名称:rapidpro,代码行数:9,代码来源:fields.py
示例7: _create_contact_batch
def _create_contact_batch(self, batch):
"""
Bulk creates a batch of contacts from flat representations
"""
for c in batch:
c["object"] = Contact(
org=c["org"],
name=c["name"],
language=c["language"],
is_stopped=c["is_stopped"],
is_blocked=c["is_blocked"],
is_active=c["is_active"],
created_by=c["user"],
created_on=c["created_on"],
modified_by=c["user"],
modified_on=c["modified_on"],
fields=c["fields_as_json"],
)
Contact.objects.bulk_create([c["object"] for c in batch])
# now that contacts have pks, bulk create the actual URN, value and group membership objects
batch_urns = []
batch_memberships = []
for c in batch:
org = c["org"]
c["urns"] = []
if c["tel"]:
c["urns"].append(
ContactURN(
org=org,
contact=c["object"],
priority=50,
scheme=TEL_SCHEME,
path=c["tel"],
identity=URN.from_tel(c["tel"]),
)
)
if c["twitter"]:
c["urns"].append(
ContactURN(
org=org,
contact=c["object"],
priority=50,
scheme=TWITTER_SCHEME,
path=c["twitter"],
identity=URN.from_twitter(c["twitter"]),
)
)
for g in c["groups"]:
batch_memberships.append(ContactGroup.contacts.through(contact=c["object"], contactgroup=g))
batch_urns += c["urns"]
ContactURN.objects.bulk_create(batch_urns)
ContactGroup.contacts.through.objects.bulk_create(batch_memberships)
开发者ID:teehamaral,项目名称:rapidpro,代码行数:57,代码来源:test_db.py
示例8: validate_urn
def validate_urn(value, strict=True):
try:
normalized = URN.normalize(value)
if strict and not URN.validate(normalized):
raise ValueError()
except ValueError:
raise serializers.ValidationError("Invalid URN: %s. Ensure phone numbers contain country codes." % value)
return normalized
开发者ID:Ilhasoft,项目名称:rapidpro,代码行数:9,代码来源:fields.py
示例9: get_object
def get_object(self, value):
# try to normalize as URN but don't blow up if it's a UUID
try:
as_urn = URN.identity(URN.normalize(value))
except ValueError:
as_urn = value
contact_ids_with_urn = list(ContactURN.objects.filter(identity=as_urn).values_list("contact_id", flat=True))
return self.get_queryset().filter(Q(uuid=value) | Q(id__in=contact_ids_with_urn)).first()
开发者ID:teehamaral,项目名称:rapidpro,代码行数:10,代码来源:fields.py
示例10: validate_urns
def validate_urns(self, value):
if value is not None:
self.parsed_urns = []
for urn in value:
try:
normalized = URN.normalize(urn)
scheme, path, query, display = URN.to_parts(normalized)
# for backwards compatibility we don't validate phone numbers here
if scheme != TEL_SCHEME and not URN.validate(normalized): # pragma: needs cover
raise ValueError()
except ValueError:
raise serializers.ValidationError("Invalid URN: '%s'" % urn)
self.parsed_urns.append(normalized)
return value
开发者ID:teehamaral,项目名称:rapidpro,代码行数:16,代码来源:serializers.py
示例11: start_call
def start_call(self, call, to, from_, status_callback):
if not settings.SEND_CALLS:
raise ValueError("SEND_CALLS set to False, skipping call start")
channel = call.channel
Contact.get_or_create(channel.org, URN.from_tel(to), channel)
# Verboice differs from Twilio in that they expect the first block of twiml up front
payload = str(Flow.handle_call(call))
# now we can post that to verboice
url = "%s?%s" % (self.endpoint, urlencode(dict(channel=self.verboice_channel, address=to)))
response = requests.post(url, data=payload, auth=self.auth).json()
if "call_id" not in response:
call.status = IVRCall.FAILED
call.save()
raise IVRException(_("Verboice connection failed."))
# store the verboice call id in our IVRCall
call.external_id = response["call_id"]
# the call was successfully sent to the IVR provider
call.status = IVRCall.WIRED
call.save()
开发者ID:teehamaral,项目名称:rapidpro,代码行数:26,代码来源:clients.py
示例12: validate
def validate(self, data):
urns = data.get("urn", [])
phones = data.get("phone", [])
contacts = data.get("contact", [])
channel = data.get("channel")
if (not urns and not phones and not contacts) or (urns and phones): # pragma: needs cover
raise serializers.ValidationError("Must provide either urns or phone or contact and not both")
if not channel:
channel = Channel.objects.filter(is_active=True, org=self.org).order_by("-last_seen").first()
if not channel: # pragma: no cover
raise serializers.ValidationError("There are no channels for this organization.")
data["channel"] = channel
if phones:
if self.org.is_anon: # pragma: needs cover
raise serializers.ValidationError("Cannot create messages for anonymous organizations")
# check our numbers for validity
country = channel.country
for urn in phones:
try:
tel, phone, query, display = URN.to_parts(urn)
normalized = phonenumbers.parse(phone, country.code)
if not phonenumbers.is_possible_number(normalized): # pragma: needs cover
raise serializers.ValidationError("Invalid phone number: '%s'" % phone)
except Exception:
raise serializers.ValidationError("Invalid phone number: '%s'" % phone)
return data
开发者ID:teehamaral,项目名称:rapidpro,代码行数:31,代码来源:serializers.py
示例13: validate_urn
def validate_urn(self, value):
urns = []
if value:
# if we have tel URNs, we may need a country to normalize by
country = self.org.get_country_code()
for urn in value:
try:
normalized = URN.normalize(urn, country)
except ValueError as e: # pragma: needs cover
raise serializers.ValidationError(str(e))
if not URN.validate(normalized, country): # pragma: needs cover
raise serializers.ValidationError("Invalid URN: '%s'" % urn)
urns.append(normalized)
return urns
开发者ID:teehamaral,项目名称:rapidpro,代码行数:17,代码来源:serializers.py
示例14: get_object
def get_object(self, value):
# try to normalize as URN but don't blow up if it's a UUID
try:
as_urn = URN.normalize(value)
except ValueError:
as_urn = value
return self.get_queryset().filter(Q(uuid=value) | Q(urns__urn=as_urn)).first()
开发者ID:Ilhasoft,项目名称:rapidpro,代码行数:8,代码来源:fields.py
示例15: to_internal_value
def to_internal_value(self, data):
if isinstance(data, str):
return [URN.from_tel(data)]
elif isinstance(data, list):
if len(data) > 100:
raise serializers.ValidationError("You can only specify up to 100 numbers at a time.")
urns = []
for phone in data:
if not isinstance(phone, str): # pragma: no cover
raise serializers.ValidationError("Invalid phone: %s" % str(phone))
urns.append(URN.from_tel(phone))
return urns
else:
raise serializers.ValidationError("Invalid phone: %s" % data)
开发者ID:teehamaral,项目名称:rapidpro,代码行数:17,代码来源:serializers.py
示例16: test_claim
def test_claim(self, mock_set_webhook, mock_get_me):
url = reverse("channels.types.telegram.claim")
self.login(self.admin)
# check that claim page URL appears on claim list page
response = self.client.get(reverse("channels.channel_claim"))
self.assertContains(response, url)
# can fetch the claim page
response = self.client.get(url)
self.assertContains(response, "Connect Telegram")
# claim with an invalid token
mock_get_me.side_effect = telegram.TelegramError("Boom")
response = self.client.post(url, {"auth_token": "invalid"})
self.assertEqual(200, response.status_code)
self.assertEqual(
"Your authentication token is invalid, please check and try again",
response.context["form"].errors["auth_token"][0],
)
user = telegram.User(123, "Rapid", True)
user.last_name = "Bot"
user.username = "rapidbot"
mock_get_me.side_effect = None
mock_get_me.return_value = user
mock_set_webhook.return_value = ""
response = self.client.post(url, {"auth_token": "184875172:BAEKbsOKAL23CXufXG4ksNV7Dq7e_1qi3j8"})
channel = Channel.objects.get(address="rapidbot")
self.assertEqual(channel.channel_type, "TG")
self.assertEqual(
channel.config,
{
"auth_token": "184875172:BAEKbsOKAL23CXufXG4ksNV7Dq7e_1qi3j8",
"callback_domain": channel.callback_domain,
},
)
self.assertRedirect(response, reverse("channels.channel_read", args=[channel.uuid]))
self.assertEqual(302, response.status_code)
response = self.client.post(url, {"auth_token": "184875172:BAEKbsOKAL23CXufXG4ksNV7Dq7e_1qi3j8"})
self.assertEqual(
"A telegram channel for this bot already exists on your account.",
response.context["form"].errors["auth_token"][0],
)
contact = self.create_contact("Telegram User", urn=URN.from_telegram("1234"))
# make sure we our telegram channel satisfies as a send channel
response = self.client.get(reverse("contacts.contact_read", args=[contact.uuid]))
send_channel = response.context["send_channel"]
self.assertIsNotNone(send_channel)
self.assertEqual(send_channel.channel_type, "TG")
开发者ID:mxabierto,项目名称:rapidpro,代码行数:57,代码来源:tests.py
示例17: validate_phone
def validate_phone(self, value):
if value:
try:
normalized = phonenumbers.parse(value, None)
if not phonenumbers.is_possible_number(normalized):
raise serializers.ValidationError("Invalid phone number: '%s'" % value)
except Exception:
raise serializers.ValidationError("Invalid phone number: '%s'" % value)
e164_number = phonenumbers.format_number(normalized, phonenumbers.PhoneNumberFormat.E164)
self.parsed_urns = [URN.from_tel(e164_number)]
return value
开发者ID:teehamaral,项目名称:rapidpro,代码行数:12,代码来源:serializers.py
示例18: reduce_event
def reduce_event(event):
new_event = copy_keys(event, {"type", "msg"})
if "msg" in new_event:
new_event["msg"] = copy_keys(event["msg"], {"text", "urn", "channel", "attachments"})
new_msg = new_event["msg"]
# legacy events are re-constructed from real messages which have their text stripped
if "text" in new_msg:
new_msg["text"] = new_msg["text"].strip()
# legacy events have absolute paths for attachments, new have relative
if "attachments" in new_msg:
abs_prefix = f"https://{settings.AWS_BUCKET_DOMAIN}/"
new_msg["attachments"] = [a.replace(abs_prefix, "") for a in new_msg["attachments"]]
# new engine events might have params on URNs that we're not interested in
if "urn" in new_msg:
scheme, path, query, fragment = URN.to_parts(new_msg["urn"])
new_msg["urn"] = URN.from_parts(scheme, path, None, fragment)
return new_event
开发者ID:mxabierto,项目名称:rapidpro,代码行数:22,代码来源:utils.py
示例19: default
def default(self, line):
"""
Sends a message as the current contact's highest priority URN
"""
urn = self.contact.get_urn()
incoming = Msg.create_incoming(None, URN.from_parts(urn.scheme, urn.path),
line, date=timezone.now(), org=self.org)
self.echo((Fore.GREEN + "[%s] " + Fore.YELLOW + ">>" + Fore.MAGENTA + " %s" + Fore.WHITE) % (urn.urn, incoming.text))
# look up any message responses
outgoing = Msg.all_messages.filter(org=self.org, pk__gt=incoming.pk, direction=OUTGOING).order_by('sent_on')
for response in outgoing:
self.echo((Fore.GREEN + "[%s] " + Fore.YELLOW + "<<" + Fore.MAGENTA + " %s" + Fore.WHITE) % (urn.urn, response.text))
开发者ID:Ebaneck,项目名称:rapidpro,代码行数:15,代码来源:msg_console.py
示例20: form_valid
def form_valid(self, *args, **kwargs):
data = self.form.cleaned_data
handled = Msg.create_incoming(data['channel'], URN.from_tel(data['urn']), data['text'],
user=self.request.user)
kwargs = self.get_form_kwargs()
kwargs['initial'] = data
next_form = TestMessageForm(**kwargs)
context = self.get_context_data()
context['handled'] = handled
context['form'] = next_form
context['responses'] = handled.responses.all()
# passing a minimal base template and a simple Context (instead of RequestContext) helps us
# minimize number of other queries, allowing us to more easily measure queries per request
context['base_template'] = 'msgs/msg_test_frame.html'
return self.render_to_response(Context(context))
开发者ID:ewheeler,项目名称:rapidpro,代码行数:18,代码来源:views.py
注:本文中的temba.contacts.models.URN类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论