Merge pull request #93 from cs50/develop

Fixes colons in strings, catches SQLAlchemy warnings

Author: Kareem Zidane

setup.py

@@ -16,5 +16,5 @@
     package_dir={"": "src"},
     packages=["cs50"],
     url="https://github.com/cs50/python-cs50",
-    version="4.0.2"
+    version="4.0.3"
 )

src/cs50/flask.py

@@ -1,7 +1,6 @@
 import logging
 
 from distutils.version import StrictVersion
-from os import getenv
 from pkg_resources import get_distribution
 
 from .cs50 import _formatException
@@ -44,7 +43,8 @@ def _execute_after(*args, **kwargs):
 
     # When behind CS50 IDE's proxy, ensure that flask.redirect doesn't redirect from HTTPS to HTTP
     # https://werkzeug.palletsprojects.com/en/0.15.x/middleware/proxy_fix/#module-werkzeug.middleware.proxy_fix
-    if getenv("C9_HOSTNAME") and not getenv("IDE_OFFLINE"):
+    from os import getenv
+    if getenv("CS50_IDE_TYPE") == "online":
         try:
             import flask
             from werkzeug.middleware.proxy_fix import ProxyFix

src/cs50/sql.py

@@ -1,17 +1,3 @@
-import datetime
-import decimal
-import importlib
-import logging
-import os
-import re
-import sqlalchemy
-import sqlite3
-import sqlparse
-import sys
-import termcolor
-import warnings
-
-
 class SQL(object):
     """Wrap SQLAlchemy to provide a simple SQL API."""
 
@@ -25,6 +11,13 @@ def __init__(self, url, **kwargs):
         http://docs.sqlalchemy.org/en/latest/dialects/index.html
         """
 
+        # Lazily import
+        import logging
+        import os
+        import re
+        import sqlalchemy
+        import sqlite3
+
         # Get logger
         self._logger = logging.getLogger("cs50")
 
@@ -74,6 +67,14 @@ def connect(dbapi_connection, connection_record):
     def execute(self, sql, *args, **kwargs):
         """Execute a SQL statement."""
 
+        # Lazily import
+        import decimal
+        import re
+        import sqlalchemy
+        import sqlparse
+        import termcolor
+        import warnings
+
         # Allow only one statement at a time, since SQLite doesn't support multiple
         # https://docs.python.org/3/library/sqlite3.html#sqlite3.Cursor.execute
         statements = sqlparse.parse(sql)
@@ -212,65 +213,82 @@ def execute(self, sql, *args, **kwargs):
                     "value" if len(keys) == 1 else "values",
                     ", ".join(keys)))
 
-        # Join tokens into statement
-        statement = "".join([str(token) for token in tokens])
+        # For SQL statements where a colon is required verbatim, as within an inline string, use a backslash to escape
+        # https://docs.sqlalchemy.org/en/13/core/sqlelement.html?highlight=text#sqlalchemy.sql.expression.text
+        for index, token in enumerate(tokens):
 
-        # Raise exceptions for warnings
-        warnings.filterwarnings("error")
+            # In string literal
+            # https://www.sqlite.org/lang_keywords.html
+            if token.ttype in [sqlparse.tokens.Literal.String, sqlparse.tokens.Literal.String.Single]:
+                token.value = re.sub("(^'|\s+):", r"\1\:", token.value)
 
-        # Prepare, execute statement
-        try:
+            # In identifier
+            # https://www.sqlite.org/lang_keywords.html
+            elif token.ttype == sqlparse.tokens.Literal.String.Symbol:
+                token.value = re.sub("(^\"|\s+):", r"\1\:", token.value)
 
-            # Execute statement
-            result = self.engine.execute(sqlalchemy.text(statement))
+        # Join tokens into statement
+        statement = "".join([str(token) for token in tokens])
 
-            # Return value
-            ret = True
-            if tokens[0].ttype == sqlparse.tokens.Keyword.DML:
-
-                # Uppercase token's value
-                value = tokens[0].value.upper()
-
-                # If SELECT, return result set as list of dict objects
-                if value == "SELECT":
-
-                    # Coerce any decimal.Decimal objects to float objects
-                    # https://groups.google.com/d/msg/sqlalchemy/0qXMYJvq8SA/oqtvMD9Uw-kJ
-                    rows = [dict(row) for row in result.fetchall()]
-                    for row in rows:
-                        for column in row:
-                            if type(row[column]) is decimal.Decimal:
-                                row[column] = float(row[column])
-                    ret = rows
-
-                # If INSERT, return primary key value for a newly inserted row
-                elif value == "INSERT":
-                    if self.engine.url.get_backend_name() in ["postgres", "postgresql"]:
-                        result = self.engine.execute("SELECT LASTVAL()")
-                        ret = result.first()[0]
-                    else:
-                        ret = result.lastrowid
-
-                # If DELETE or UPDATE, return number of rows matched
-                elif value in ["DELETE", "UPDATE"]:
-                    ret = result.rowcount
-
-        # If constraint violated, return None
-        except sqlalchemy.exc.IntegrityError:
-            self._logger.debug(termcolor.colored(statement, "yellow"))
-            return None
-
-        # If user errror
-        except sqlalchemy.exc.OperationalError as e:
-            self._logger.debug(termcolor.colored(statement, "red"))
-            e = RuntimeError(_parse_exception(e))
-            e.__cause__ = None
-            raise e
+        # Catch SQLAlchemy warnings
+        with warnings.catch_warnings():
+
+            # Raise exceptions for warnings
+            warnings.simplefilter("error")
+
+            # Prepare, execute statement
+            try:
+
+                # Execute statement
+                result = self.engine.execute(sqlalchemy.text(statement))
+
+                # Return value
+                ret = True
+                if tokens[0].ttype == sqlparse.tokens.Keyword.DML:
+
+                    # Uppercase token's value
+                    value = tokens[0].value.upper()
+
+                    # If SELECT, return result set as list of dict objects
+                    if value == "SELECT":
+
+                        # Coerce any decimal.Decimal objects to float objects
+                        # https://groups.google.com/d/msg/sqlalchemy/0qXMYJvq8SA/oqtvMD9Uw-kJ
+                        rows = [dict(row) for row in result.fetchall()]
+                        for row in rows:
+                            for column in row:
+                                if type(row[column]) is decimal.Decimal:
+                                    row[column] = float(row[column])
+                        ret = rows
+
+                    # If INSERT, return primary key value for a newly inserted row
+                    elif value == "INSERT":
+                        if self.engine.url.get_backend_name() in ["postgres", "postgresql"]:
+                            result = self.engine.execute("SELECT LASTVAL()")
+                            ret = result.first()[0]
+                        else:
+                            ret = result.lastrowid
+
+                    # If DELETE or UPDATE, return number of rows matched
+                    elif value in ["DELETE", "UPDATE"]:
+                        ret = result.rowcount
+
+            # If constraint violated, return None
+            except sqlalchemy.exc.IntegrityError:
+                self._logger.debug(termcolor.colored(statement, "yellow"))
+                return None
+
+            # If user errror
+            except sqlalchemy.exc.OperationalError as e:
+                self._logger.debug(termcolor.colored(statement, "red"))
+                e = RuntimeError(_parse_exception(e))
+                e.__cause__ = None
+                raise e
 
-        # Return value
-        else:
-            self._logger.debug(termcolor.colored(statement, "green"))
-            return ret
+            # Return value
+            else:
+                self._logger.debug(termcolor.colored(statement, "green"))
+                return ret
 
     def _escape(self, value):
         """
@@ -279,8 +297,15 @@ def _escape(self, value):
         https://docs.sqlalchemy.org/en/latest/core/type_api.html#sqlalchemy.types.TypeEngine.literal_processor
         """
 
+        # Lazily import
+        import sqlparse
+
         def __escape(value):
 
+            # Lazily import
+            import datetime
+            import sqlalchemy
+
             # bool
             if type(value) is bool:
                 return sqlparse.sql.Token(
@@ -349,6 +374,9 @@ def __escape(value):
 def _parse_exception(e):
     """Parses an exception, returns its message."""
 
+    # Lazily import
+    import re
+
     # MySQL
     matches = re.search(r"^\(_mysql_exceptions\.OperationalError\) \(\d+, \"(.+)\"\)$", str(e))
     if matches:
@@ -371,6 +399,10 @@ def _parse_exception(e):
 def _parse_placeholder(token):
     """Infers paramstyle, name from sqlparse.tokens.Name.Placeholder."""
 
+    # Lazily load
+    import re
+    import sqlparse
+
     # Validate token
     if not isinstance(token, sqlparse.sql.Token) or token.ttype != sqlparse.tokens.Name.Placeholder:
         raise TypeError()

tests/python.py

@@ -4,5 +4,5 @@
 
 import cs50
 
-i = cs50.get_int()
-print(i)
+i = cs50.get_int("Input:  ")
+print(f"Output: {i}")

tests/sql.py

@@ -75,6 +75,24 @@ def test_update_returns_affected_rows(self):
         self.assertEqual(self.db.execute("UPDATE cs50 SET val = 'foo' WHERE id > 1"), 2)
         self.assertEqual(self.db.execute("UPDATE cs50 SET val = 'foo' WHERE id = -50"), 0)
 
+    def test_string_literal_with_colon(self):
+        rows = [
+            {"id": 1, "val": ":foo"},
+            {"id": 2, "val": "foo:bar"},
+            {"id": 3, "val": "  :baz"},
+            {"id": 3, "val": ":bar :baz"},
+            {"id": 3, "val": "  :bar :baz"}
+        ]
+        for row in rows:
+            self.db.execute("INSERT INTO cs50(val) VALUES(:val)", val=row["val"])
+
+        self.assertEqual(self.db.execute("SELECT val FROM cs50 WHERE val = ':foo'"), [{"val": ":foo"}])
+        self.assertEqual(self.db.execute("SELECT val FROM cs50 WHERE val = ':bar'"), [])
+        self.assertEqual(self.db.execute("SELECT val FROM cs50 WHERE val = 'foo:bar'"), [{"val": "foo:bar"}])
+        self.assertEqual(self.db.execute("SELECT val FROM cs50 WHERE val = '  :baz'"), [{"val": "  :baz"}])
+        self.assertEqual(self.db.execute("SELECT val FROM cs50 WHERE val = ':bar :baz'"), [{"val": ":bar :baz"}])
+        self.assertEqual(self.db.execute("SELECT val FROM cs50 WHERE val = '  :bar :baz'"), [{"val": "  :bar :baz"}])
+
     def tearDown(self):
         self.db.execute("DROP TABLE cs50")
 

tests/sqlite.py

@@ -31,6 +31,8 @@
 db.execute("SELECT * FROM Employee WHERE FirstName = :1 AND LastName = :2", ["Andrew", "Adams"])
 db.execute("SELECT * FROM Employee WHERE FirstName = :1 AND LastName = :2", ("Andrew", "Adams"))
 
+db.execute("SELECT * FROM Employee WHERE FirstName = ':Andrew :Adams'")
+
 db.execute("SELECT * FROM Employee WHERE FirstName = :first AND LastName = :last", first="Andrew", last="Adams")
 db.execute("SELECT * FROM Employee WHERE FirstName = :first AND LastName = :last", {"first": "Andrew", "last": "Adams"})