Move code that escapes special chars to helper function and use it

in query get_users_by_partial_email

app/dao/notifications_dao.py

@@ -22,7 +22,7 @@ from sqlalchemy.sql import functions
 from notifications_utils.international_billing_rates import INTERNATIONAL_BILLING_RATES
 
 from app import db, create_uuid
-from app.utils import midnight_n_days_ago
+from app.utils import midnight_n_days_ago, escape_special_characters
 from app.errors import InvalidRequest
 from app.models import (
     Notification,
@@ -452,11 +452,7 @@ def dao_get_notifications_by_to_field(service_id, search_term, notification_type
     else:
         raise InvalidRequest("Only email and SMS can use search by recipient", 400)
 
-    for special_character in ('\\', '_', '%', '/'):
-        normalised = normalised.replace(
-            special_character,
-            '\{}'.format(special_character)
-        )
+    normalised = escape_special_characters(normalised)
 
     filters = [
         Notification.service_id == service_id,

app/dao/users_dao.py

@@ -6,6 +6,7 @@ from sqlalchemy.orm import joinedload
 
 from app import db
 from app.models import (User, VerifyCode)
+from app.utils import escape_special_characters
 
 
 def _remove_values_for_keys_if_present(dict, keys):
@@ -98,7 +99,8 @@ def get_user_by_email(email):
 
 
 def get_users_by_partial_email(email):
-    return User.query.filter(User.email_address.ilike("\%{}\%".format(email))).all()
+    email = escape_special_characters(email)
+    return User.query.filter(User.email_address.ilike("%{}%".format(email))).all()
 
 
 def increment_failed_login_count(user):