validate_edit_message() - Code Metrics - Inspection of "More tornado complexity refactoring" - Deathangel908/pychat - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 8205ad...0b3f02 )

by Andrew

created 2017-07-02 20:30 UTC

validate_edit_message() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	4
c	1
b	0
f	0
dl	0
loc	7
rs	9.2

import base64
import json
import logging
import re
import sys
from io import BytesIO

import requests
from django.contrib.auth import get_user_model
from django.core.exceptions import ValidationError
from django.core.files.uploadedfile import InMemoryUploadedFile
from django.core.mail import send_mail
from django.core.validators import validate_email
from django.db import IntegrityError
from django.db import connection, OperationalError, InterfaceError
from django.template import RequestContext
from django.template.loader import render_to_string
from django.utils.safestring import mark_safe

from chat import local
from chat import settings
from chat.models import Room, get_milliseconds
from chat.models import User
from chat.models import UserProfile, Verification, RoomUsers, IpAddress
from chat.py2_3 import urlopen
from chat.settings import ISSUES_REPORT_LINK, SITE_PROTOCOL, ALL_ROOM_ID
from chat.tornado.constants import RedisPrefix
from chat.tornado.constants import VarNames

USERNAME_REGEX = str(settings.MAX_USERNAME_LENGTH).join(['^[a-zA-Z-_0-9]{1,', '}$'])
API_URL = getattr(settings, "IP_API_URL", None)
RECAPTCHA_SECRET_KEY = getattr(settings, "RECAPTCHA_SECRET_KEY", None)
GOOGLE_OAUTH_2_CLIENT_ID = getattr(settings, "GOOGLE_OAUTH_2_CLIENT_ID", None)
GOOGLE_OAUTH_2_HOST = getattr(settings, "GOOGLE_OAUTH_2_HOST", None)
FACEBOOK_ACCESS_TOKEN = getattr(settings, "FACEBOOK_ACCESS_TOKEN", None)

logger = logging.getLogger(__name__)


def is_blank(check_str):
	if check_str and check_str.strip():
		return False
	else:
		return True


def get_users_in_current_user_rooms(user_id):
	user_rooms = Room.objects.filter(users__id=user_id, disabled=False).values('id', 'name')
	res = {room['id']: {
			VarNames.ROOM_NAME: room['name'],
			VarNames.ROOM_USERS: {}
		} for room in user_rooms}
	room_ids = (room_id for room_id in res)
	rooms_users = User.objects.filter(rooms__in=room_ids).values('id', 'username', 'sex', 'rooms__id')
	for user in rooms_users:
		dict = res[user['rooms__id']][VarNames.ROOM_USERS]
		dict[user['id']] = RedisPrefix.set_js_user_structure(user['username'], user['sex'])
	return res


def get_or_create_room(channels, room_id, user_id):
	if room_id not in channels:
		raise ValidationError("Access denied, only allowed for channels {}".format(channels))
	room = do_db(Room.objects.get, id=room_id)
	if room.is_private:
		raise ValidationError("You can't add users to direct room, create a new room instead")
	try:
		Room.users.through.objects.create(room_id=room_id, user_id=user_id)
	except IntegrityError:
		raise ValidationError("User is already in channel")
	return room


def update_room(room_id, disabled):
	if not disabled:
		raise ValidationError('This room already exist')
	else:
		Room.objects.filter(id=room_id).update(disabled=False)


def create_room_users(me, other):
	room = Room()
	room.save()
	room_id = room.id
	if me == other:
		RoomUsers(user_id=me, room_id=room_id).save()
	else:
		RoomUsers.objects.bulk_create([
			RoomUsers(user_id=other, room_id=room_id),
			RoomUsers(user_id=me, room_id=room_id),
		])
	return room_id


def validate_edit_message(self_id, message):
	if message.sender_id != self_id:
		raise ValidationError("You can only edit your messages")
	if message.time + 600000 < get_milliseconds():
		raise ValidationError("You can only edit messages that were send not more than 10 min ago")
	if message.deleted:
		raise ValidationError("Already deleted")


def do_db(callback, *args, **kwargs):
	try:
		return callback(*args, **kwargs)
	except (OperationalError, InterfaceError) as e:
		if 'MySQL server has gone away' in str(e):
			logger.warning('%s, reconnecting' % e)
			connection.close()
			return callback(*args, **kwargs)
		else:
			raise e


def execute_query(query, *args, **kwargs):
	cursor = connection.cursor()
	cursor.execute(query, *args, **kwargs)
	desc = cursor.description
	return [
		dict(zip([col[0] for col in desc], row))
		for row in cursor.fetchall()
	]


def hide_fields(post, fields, huge=False, fill_with='****'):
	"""
	:param post: Object that will be copied
	:type post: QueryDict
	:param fields: fields that will be removed
	:param fill_with: replace characters for hidden fields
	:param huge: if true object will be cloned and then fields will be removed
	:return: a shallow copy of dictionary without specified fields
	"""
	if not huge:
		# hide *fields in shallow copy
		res = post.copy()
		for field in fields:
			if field in post:  # check if field was absent
				res[field] = fill_with
	else:
		# copy everything but *fields
		res = {}
		for field in post:
			# _______________________if this is field to remove
			res[field] = post[field] if field not in fields else fill_with
	return res


def check_password(password):
	"""
	Checks if password is secure
	:raises ValidationError exception if password is not valid
	"""
	if is_blank(password):
		raise ValidationError("password can't be empty")
	if not re.match(u'^\S.+\S$', password):
		raise ValidationError("password should be at least 3 symbols")


def check_email(email):
	"""
	:raises ValidationError if specified email is registered or not valid
	"""
	if not email:
		return
	try:
		validate_email(email)
		# theoretically can throw returning 'more than 1' error
		UserProfile.objects.get(email=email)
		raise ValidationError('Email {} is already used'.format(email))
	except User.DoesNotExist:
		pass


def check_user(username):
	"""
	Checks if specified username is free to register
	:type username str
	:raises ValidationError exception if username is not valid
	"""
	# Skip javascript validation, only summary message
	if is_blank(username):
		raise ValidationError("Username can't be empty")
	if not re.match(USERNAME_REGEX, username):
		raise ValidationError("Username {} doesn't match regex {}".format(username, USERNAME_REGEX))
	try:
		# theoretically can throw returning 'more than 1' error
		User.objects.get(username=username)
		raise ValidationError("Username {} is already used. Please select another one".format(username))
	except User.DoesNotExist:
		pass


def get_client_ip(request):
	x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
	return x_forwarded_for.split(',')[-1].strip() if x_forwarded_for else request.META.get('REMOTE_ADDR')


def check_captcha(request):
	"""
	:type request: WSGIRequest
	:raises ValidationError: if captcha is not valid or not set
	If RECAPTCHA_SECRET_KEY is enabled in settings validates request with it
	"""
	if not RECAPTCHA_SECRET_KEY:
		logger.debug('Skipping captcha validation')
		return
	try:
		captcha_rs = request.POST.get('g-recaptcha-response')
		url = "https://www.google.com/recaptcha/api/siteverify"
		params = {
			'secret': RECAPTCHA_SECRET_KEY,
			'response': captcha_rs,
			'remoteip': local.client_ip
		}
		raw_response = requests.post(url, params=params, verify=True)
		response = raw_response.json()
		if not response.get('success', False):
			logger.debug('Captcha is NOT valid, response: %s', raw_response)
			raise ValidationError(
				response['error-codes'] if response.get('error-codes', None) else 'This captcha already used')
		logger.debug('Captcha is valid, response: %s', raw_response)
	except Exception as e:
		raise ValidationError('Unable to check captcha because {}'.format(e))


def send_sign_up_email(user, site_address, request):
	if user.email is not None:
		verification = Verification(user=user, type_enum=Verification.TypeChoices.register)
		verification.save()
		user.email_verification = verification
		user.save(update_fields=['email_verification'])
		link = "{}://{}/confirm_email?token={}".format(SITE_PROTOCOL, site_address, verification.token)
		text = ('Hi {}, you have registered pychat'
				  '\nTo complete your registration please click on the url bellow: {}'
				  '\n\nIf you find any bugs or propositions you can post them {}').format(
			user.username, link, ISSUES_REPORT_LINK)
		start_message = mark_safe((
			"You have registered in <b>Pychat</b>. If you find any bugs or propositions you can post them"
			" <a href='{}'>here</a>. To complete your registration please click on the link below.").format(
				ISSUES_REPORT_LINK))
		context = {
			'username': user.username,
			'magicLink': link,
			'btnText': "CONFIRM SIGN UP",
			'greetings': start_message
		}
		html_message = render_to_string('sign_up_email.html', context, context_instance=RequestContext(request))
		logger.info('Sending verification email to userId %s (email %s)', user.id, user.email)
		send_mail("Confirm chat registration", text, site_address, [user.email, ], html_message=html_message)
		logger.info('Email %s has been sent', user.email)


def send_reset_password_email(request, user_profile, verification):
	link = "{}://{}/restore_password?token={}".format(SITE_PROTOCOL, request.get_host(), verification.token)
	message = "{},\n" \
				 "You requested to change a password on site {}.\n" \
				 "To proceed click on the link {}\n" \
				 "If you didn't request the password change just ignore this mail" \
		.format(user_profile.username, request.get_host(), link)
	ip_info = get_or_create_ip(get_client_ip(request), logger)
	start_message = mark_safe(
		"You have requested to send you a magic link to quickly restore password to <b>Pychat</b>. "
		"If it wasn't you, you can safely ignore this email")
	context = {
		'username': user_profile.username,
		'magicLink': link,
		'ipInfo': ip_info.info,
		'ip': ip_info.ip,
		'btnText': "CHANGE PASSWORD",
		'timeCreated': verification.time,
		'greetings': start_message
	}
	html_message = render_to_string('reset_pass_email.html', context, context_instance=RequestContext(request))
	send_mail("Pychat: restore password", message, request.get_host(), (user_profile.email,), fail_silently=False,
				 html_message=html_message)


def extract_photo(image_base64, filename=None):
	base64_type_data = re.search(r'data:(\w+/(\w+));base64,(.*)$', image_base64)
	logger.debug('Parsing base64 image')
	image_data = base64_type_data.group(3)
	f = BytesIO(base64.b64decode(image_data))
	content_type = base64_type_data.group(1)
	name = filename or ".{}".format(base64_type_data.group(2))
	logger.debug('Base64 filename extension %s, content_type %s', name, content_type)
	image = InMemoryUploadedFile(
		f,
		field_name='photo',
		name=name,
		content_type=content_type,
		size=sys.getsizeof(f),
		charset=None)
	return image


def get_max_key(dictionary):
	return max(dictionary.keys()) if dictionary else None


def create_user_model(user):
	user.save()
	RoomUsers(user_id=user.id, room_id=ALL_ROOM_ID).save()
	logger.info('Signed up new user %s, subscribed for channels with id %d', user, ALL_ROOM_ID)
	return user


def get_or_create_ip(ip, logger):
	"""

	@param ip: ip to fetch info from
	@param logger initialized logger:
	@type IpAddress
	"""
	try:
		ip_address = IpAddress.objects.get(ip=ip)
	except IpAddress.DoesNotExist:
		try:
			if not API_URL:
				raise Exception('api url is absent')
			logger.debug("Creating ip record %s", ip)
			f = urlopen(API_URL % ip)
			raw_response = f.read().decode("utf-8")
			response = json.loads(raw_response)
			if response['status'] != "success":
				raise Exception("Creating iprecord failed, server responded: %s" % raw_response)
			ip_address = IpAddress.objects.create(
				ip=ip,
				isp=response['isp'],
				country=response['country'],
				region=response['regionName'],
				city=response['city'],
				country_code=response['countryCode']
			)
		except Exception as e:
			logger.error("Error while creating ip with country info, because %s", e)
			ip_address = IpAddress.objects.create(ip=ip)
	return ip_address


class EmailOrUsernameModelBackend(object):
	"""
	This is a ModelBacked that allows authentication with either a username or an email address.
	"""

	def authenticate(self, username=None, password=None):
		try:
			if '@' in username:
				user = UserProfile.objects.get(email=username)
			else:
				user = UserProfile.objects.get(username=username)
			if user.check_password(password):
				return user
		except User.DoesNotExist:
			return None

	def get_user(self, username):
		try:
			return get_user_model().objects.get(pk=username)
		except get_user_model().DoesNotExist:
			return None


1			import base64
2			import json
3			import logging
4			import re
5			import sys
6			from io import BytesIO
7
8			import requests
9			from django.contrib.auth import get_user_model
10			from django.core.exceptions import ValidationError
11			from django.core.files.uploadedfile import InMemoryUploadedFile
12			from django.core.mail import send_mail
13			from django.core.validators import validate_email
14			from django.db import IntegrityError
15			from django.db import connection, OperationalError, InterfaceError
16			from django.template import RequestContext
17			from django.template.loader import render_to_string
18			from django.utils.safestring import mark_safe
19
20			from chat import local
21			from chat import settings
22			from chat.models import Room, get_milliseconds
23			from chat.models import User
24			from chat.models import UserProfile, Verification, RoomUsers, IpAddress
25			from chat.py2_3 import urlopen
26			from chat.settings import ISSUES_REPORT_LINK, SITE_PROTOCOL, ALL_ROOM_ID
27			from chat.tornado.constants import RedisPrefix
28			from chat.tornado.constants import VarNames
29
30			USERNAME_REGEX = str(settings.MAX_USERNAME_LENGTH).join(['^[a-zA-Z-_0-9]{1,', '}$'])
31			API_URL = getattr(settings, "IP_API_URL", None)
32			RECAPTCHA_SECRET_KEY = getattr(settings, "RECAPTCHA_SECRET_KEY", None)
33			GOOGLE_OAUTH_2_CLIENT_ID = getattr(settings, "GOOGLE_OAUTH_2_CLIENT_ID", None)
34			GOOGLE_OAUTH_2_HOST = getattr(settings, "GOOGLE_OAUTH_2_HOST", None)
35			FACEBOOK_ACCESS_TOKEN = getattr(settings, "FACEBOOK_ACCESS_TOKEN", None)
36
37			logger = logging.getLogger(__name__)
38
39
40			def is_blank(check_str):
41			if check_str and check_str.strip():
42			return False
43			else:
44			return True
45
46
47			def get_users_in_current_user_rooms(user_id):
48			user_rooms = Room.objects.filter(users__id=user_id, disabled=False).values('id', 'name')
49			res = {room['id']: {
50			VarNames.ROOM_NAME: room['name'],
51			VarNames.ROOM_USERS: {}
52			} for room in user_rooms}
53			room_ids = (room_id for room_id in res)
54			rooms_users = User.objects.filter(rooms__in=room_ids).values('id', 'username', 'sex', 'rooms__id')
55			for user in rooms_users:
56			dict = res[user['rooms__id']][VarNames.ROOM_USERS]
57			dict[user['id']] = RedisPrefix.set_js_user_structure(user['username'], user['sex'])
58			return res
59
60
61			def get_or_create_room(channels, room_id, user_id):
62			if room_id not in channels:
63			raise ValidationError("Access denied, only allowed for channels {}".format(channels))
64			room = do_db(Room.objects.get, id=room_id)
65			if room.is_private:
66			raise ValidationError("You can't add users to direct room, create a new room instead")
67			try:
68			Room.users.through.objects.create(room_id=room_id, user_id=user_id)
69			except IntegrityError:
70			raise ValidationError("User is already in channel")
71			return room
72
73
74			def update_room(room_id, disabled):
75			if not disabled:
76			raise ValidationError('This room already exist')
77			else:
78			Room.objects.filter(id=room_id).update(disabled=False)
79
80
81			def create_room_users(me, other):
82			room = Room()
83			room.save()
84			room_id = room.id
85			if me == other:
86			RoomUsers(user_id=me, room_id=room_id).save()
87			else:
88			RoomUsers.objects.bulk_create([
89			RoomUsers(user_id=other, room_id=room_id),
90			RoomUsers(user_id=me, room_id=room_id),
91			])
92			return room_id
93
94
95			def validate_edit_message(self_id, message):
96			if message.sender_id != self_id:
97			raise ValidationError("You can only edit your messages")
98			if message.time + 600000 < get_milliseconds():
99			raise ValidationError("You can only edit messages that were send not more than 10 min ago")
100			if message.deleted:
101			raise ValidationError("Already deleted")
102
103
104			def do_db(callback, args, *kwargs):
105			try:
106			return callback(args, *kwargs)
107			except (OperationalError, InterfaceError) as e:
108			if 'MySQL server has gone away' in str(e):
109			logger.warning('%s, reconnecting' % e)
110			connection.close()
111			return callback(args, *kwargs)
112			else:
113			raise e
114
115
116			def execute_query(query, args, *kwargs):
117			cursor = connection.cursor()
118			cursor.execute(query, args, *kwargs)
119			desc = cursor.description
120			return [
121			dict(zip([col[0] for col in desc], row))
122			for row in cursor.fetchall()
123			]
124
125
126			def hide_fields(post, fields, huge=False, fill_with='****'):
127			"""
128			:param post: Object that will be copied
129			:type post: QueryDict
130			:param fields: fields that will be removed
131			:param fill_with: replace characters for hidden fields
132			:param huge: if true object will be cloned and then fields will be removed
133			:return: a shallow copy of dictionary without specified fields
134			"""
135			if not huge:
136			# hide *fields in shallow copy
137			res = post.copy()
138			for field in fields:
139			if field in post: # check if field was absent
140			res[field] = fill_with
141			else:
142			# copy everything but *fields
143			res = {}
144			for field in post:
145			# _______________________if this is field to remove
146			res[field] = post[field] if field not in fields else fill_with
147			return res
148
149
150			def check_password(password):
151			"""
152			Checks if password is secure
153			:raises ValidationError exception if password is not valid
154			"""
155			if is_blank(password):
156			raise ValidationError("password can't be empty")
157			if not re.match(u'^\S.+\S$', password):
158			raise ValidationError("password should be at least 3 symbols")
159
160
161			def check_email(email):
162			"""
163			:raises ValidationError if specified email is registered or not valid
164			"""
165			if not email:
166			return
167			try:
168			validate_email(email)
169			# theoretically can throw returning 'more than 1' error
170			UserProfile.objects.get(email=email)
171			raise ValidationError('Email {} is already used'.format(email))
172			except User.DoesNotExist:
173			pass
174
175
176			def check_user(username):
177			"""
178			Checks if specified username is free to register
179			:type username str
180			:raises ValidationError exception if username is not valid
181			"""
182			# Skip javascript validation, only summary message
183			if is_blank(username):
184			raise ValidationError("Username can't be empty")
185			if not re.match(USERNAME_REGEX, username):
186			raise ValidationError("Username {} doesn't match regex {}".format(username, USERNAME_REGEX))
187			try:
188			# theoretically can throw returning 'more than 1' error
189			User.objects.get(username=username)
190			raise ValidationError("Username {} is already used. Please select another one".format(username))
191			except User.DoesNotExist:
192			pass
193
194
195			def get_client_ip(request):
196			x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
197			return x_forwarded_for.split(',')[-1].strip() if x_forwarded_for else request.META.get('REMOTE_ADDR')
198
199
200			def check_captcha(request):
201			"""
202			:type request: WSGIRequest
203			:raises ValidationError: if captcha is not valid or not set
204			If RECAPTCHA_SECRET_KEY is enabled in settings validates request with it
205			"""
206			if not RECAPTCHA_SECRET_KEY:
207			logger.debug('Skipping captcha validation')
208			return
209			try:
210			captcha_rs = request.POST.get('g-recaptcha-response')
211			url = "https://www.google.com/recaptcha/api/siteverify"
212			params = {
213			'secret': RECAPTCHA_SECRET_KEY,
214			'response': captcha_rs,
215			'remoteip': local.client_ip
216			}
217			raw_response = requests.post(url, params=params, verify=True)
218			response = raw_response.json()
219			if not response.get('success', False):
220			logger.debug('Captcha is NOT valid, response: %s', raw_response)
221			raise ValidationError(
222			response['error-codes'] if response.get('error-codes', None) else 'This captcha already used')
223			logger.debug('Captcha is valid, response: %s', raw_response)
224			except Exception as e:
225			raise ValidationError('Unable to check captcha because {}'.format(e))
226
227
228			def send_sign_up_email(user, site_address, request):
229			if user.email is not None:
230			verification = Verification(user=user, type_enum=Verification.TypeChoices.register)
231			verification.save()
232			user.email_verification = verification
233			user.save(update_fields=['email_verification'])
234			link = "{}://{}/confirm_email?token={}".format(SITE_PROTOCOL, site_address, verification.token)
235			text = ('Hi {}, you have registered pychat'
236			'\nTo complete your registration please click on the url bellow: {}'
237			'\n\nIf you find any bugs or propositions you can post them {}').format(
238			user.username, link, ISSUES_REPORT_LINK)
239			start_message = mark_safe((
240			"You have registered in <b>Pychat</b>. If you find any bugs or propositions you can post them"
241			" <a href='{}'>here</a>. To complete your registration please click on the link below.").format(
242			ISSUES_REPORT_LINK))
243			context = {
244			'username': user.username,
245			'magicLink': link,
246			'btnText': "CONFIRM SIGN UP",
247			'greetings': start_message
248			}
249			html_message = render_to_string('sign_up_email.html', context, context_instance=RequestContext(request))
250			logger.info('Sending verification email to userId %s (email %s)', user.id, user.email)
251			send_mail("Confirm chat registration", text, site_address, [user.email, ], html_message=html_message)
252			logger.info('Email %s has been sent', user.email)
253
254
255			def send_reset_password_email(request, user_profile, verification):
256			link = "{}://{}/restore_password?token={}".format(SITE_PROTOCOL, request.get_host(), verification.token)
257			message = "{},\n" \
258			"You requested to change a password on site {}.\n" \
259			"To proceed click on the link {}\n" \
260			"If you didn't request the password change just ignore this mail" \
261			.format(user_profile.username, request.get_host(), link)
262			ip_info = get_or_create_ip(get_client_ip(request), logger)
263			start_message = mark_safe(
264			"You have requested to send you a magic link to quickly restore password to <b>Pychat</b>. "
265			"If it wasn't you, you can safely ignore this email")
266			context = {
267			'username': user_profile.username,
268			'magicLink': link,
269			'ipInfo': ip_info.info,
270			'ip': ip_info.ip,
271			'btnText': "CHANGE PASSWORD",
272			'timeCreated': verification.time,
273			'greetings': start_message
274			}
275			html_message = render_to_string('reset_pass_email.html', context, context_instance=RequestContext(request))
276			send_mail("Pychat: restore password", message, request.get_host(), (user_profile.email,), fail_silently=False,
277			html_message=html_message)
278
279
280			def extract_photo(image_base64, filename=None):
281			base64_type_data = re.search(r'data:(\w+/(\w+));base64,(.*)$', image_base64)
282			logger.debug('Parsing base64 image')
283			image_data = base64_type_data.group(3)
284			f = BytesIO(base64.b64decode(image_data))
285			content_type = base64_type_data.group(1)
286			name = filename or ".{}".format(base64_type_data.group(2))
287			logger.debug('Base64 filename extension %s, content_type %s', name, content_type)
288			image = InMemoryUploadedFile(
289			f,
290			field_name='photo',
291			name=name,
292			content_type=content_type,
293			size=sys.getsizeof(f),
294			charset=None)
295			return image
296
297
298			def get_max_key(dictionary):
299			return max(dictionary.keys()) if dictionary else None
300
301
302			def create_user_model(user):
303			user.save()
304			RoomUsers(user_id=user.id, room_id=ALL_ROOM_ID).save()
305			logger.info('Signed up new user %s, subscribed for channels with id %d', user, ALL_ROOM_ID)
306			return user
307
308
309			def get_or_create_ip(ip, logger):
310			"""
311
312			@param ip: ip to fetch info from
313			@param logger initialized logger:
314			@type IpAddress
315			"""
316			try:
317			ip_address = IpAddress.objects.get(ip=ip)
318			except IpAddress.DoesNotExist:
319			try:
320			if not API_URL:
321			raise Exception('api url is absent')
322			logger.debug("Creating ip record %s", ip)
323			f = urlopen(API_URL % ip)
324			raw_response = f.read().decode("utf-8")
325			response = json.loads(raw_response)
326			if response['status'] != "success":
327			raise Exception("Creating iprecord failed, server responded: %s" % raw_response)
328			ip_address = IpAddress.objects.create(
329			ip=ip,
330			isp=response['isp'],
331			country=response['country'],
332			region=response['regionName'],
333			city=response['city'],
334			country_code=response['countryCode']
335			)
336			except Exception as e:
337			logger.error("Error while creating ip with country info, because %s", e)
338			ip_address = IpAddress.objects.create(ip=ip)
339			return ip_address
340
341
342			class EmailOrUsernameModelBackend(object):
343			"""
344			This is a ModelBacked that allows authentication with either a username or an email address.
345			"""
346
347			def authenticate(self, username=None, password=None):
348			try:
349			if '@' in username:
350			user = UserProfile.objects.get(email=username)
351			else:
352			user = UserProfile.objects.get(username=username)
353			if user.check_password(password):
354			return user
355			except User.DoesNotExist:
356			return None
357
358			def get_user(self, username):
359			try:
360			return get_user_model().objects.get(pk=username)
361			except get_user_model().DoesNotExist:
362			return None
363

Deathangel908 / pychat

Push — master ( 8205ad...0b3f02 )

validate_edit_message() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like