refactor, fix scoped session

Author: Kareem Zidane

setup.py

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

src/cs50/__init__.py

@@ -1,20 +1,6 @@
-import logging
-import os
-import sys
+from ._logger import _setup_logger
+_setup_logger()
 
-
-# Disable cs50 logger by default
-logging.getLogger("cs50").disabled = True
-
-# Import cs50_*
-from .cs50 import get_char, get_float, get_int, get_string
-try:
-    from .cs50 import get_long
-except ImportError:
-    pass
-
-# Hook into flask importing
+from .cs50 import get_float, get_int, get_string
 from . import flask
-
-# Wrap SQLAlchemy
 from .sql import SQL

src/cs50/_logger.py

@@ -0,0 +1,48 @@
+import logging
+import os.path
+import re
+import sys
+import traceback
+
+import termcolor
+
+
+def _setup_logger():
+    _logger = logging.getLogger("cs50")
+    _logger.disabled = True
+    _logger.setLevel(logging.DEBUG)
+
+    # Log messages once
+    _logger.propagate = False
+
+    handler = logging.StreamHandler()
+    handler.setLevel(logging.DEBUG)
+
+    formatter = logging.Formatter("%(levelname)s: %(message)s")
+    formatter.formatException = lambda exc_info: _formatException(*exc_info)
+    handler.setFormatter(formatter)
+    _logger.addHandler(handler)
+
+
+def _formatException(type, value, tb):
+    """
+    Format traceback, darkening entries from global site-packages directories
+    and user-specific site-packages directory.
+    https://stackoverflow.com/a/46071447/5156190
+    """
+
+    # Absolute paths to site-packages
+    packages = tuple(os.path.join(os.path.abspath(p), "") for p in sys.path[1:])
+
+    # Highlight lines not referring to files in site-packages
+    lines = []
+    for line in traceback.format_exception(type, value, tb):
+        matches = re.search(r"^  File \"([^\"]+)\", line \d+, in .+", line)
+        if matches and matches.group(1).startswith(packages):
+            lines += line
+        else:
+            matches = re.search(r"^(\s*)(.*?)(\s*)$", line, re.DOTALL)
+            lines.append(matches.group(1) + termcolor.colored(matches.group(2), "yellow") + matches.group(3))
+    return "".join(lines).rstrip()
+
+

src/cs50/_session.py

@@ -0,0 +1,80 @@
+import os
+
+import sqlalchemy
+import sqlalchemy.orm
+import sqlite3
+
+class Session:
+    def __init__(self, url, **engine_kwargs):
+        self._url = url
+        if _is_sqlite_url(self._url):
+            _assert_sqlite_file_exists(self._url)
+
+        self._engine = _create_engine(self._url, **engine_kwargs)
+        self._is_postgres = self._engine.url.get_backend_name() in {"postgres", "postgresql"}
+        _setup_on_connect(self._engine)
+        self._session = _create_scoped_session(self._engine)
+
+
+    def is_postgres(self):
+        return self._is_postgres
+
+
+    def execute(self, statement):
+        return self._session.execute(sqlalchemy.text(str(statement)))
+
+
+    def __getattr__(self, attr):
+        return getattr(self._session, attr)
+
+
+def _is_sqlite_url(url):
+    return url.startswith("sqlite:///")
+
+
+def _assert_sqlite_file_exists(url):
+    path = url[len("sqlite:///"):]
+    if not os.path.exists(path):
+        raise RuntimeError(f"does not exist: {path}")
+    if not os.path.isfile(path):
+        raise RuntimeError(f"not a file: {path}")
+
+
+def _create_engine(url, **kwargs):
+    try:
+        engine = sqlalchemy.create_engine(url, **kwargs)
+    except sqlalchemy.exc.ArgumentError:
+        raise RuntimeError(f"invalid URL: {url}") from None
+
+    engine.execution_options(autocommit=False)
+    return engine
+
+
+def _setup_on_connect(engine):
+    def connect(dbapi_connection, _):
+        _disable_auto_begin_commit(dbapi_connection)
+        if _is_sqlite_connection(dbapi_connection):
+            _enable_sqlite_foreign_key_constraints(dbapi_connection)
+
+    sqlalchemy.event.listen(engine, "connect", connect)
+
+
+def _create_scoped_session(engine):
+    session_factory = sqlalchemy.orm.sessionmaker(bind=engine)
+    return sqlalchemy.orm.scoping.scoped_session(session_factory)
+
+
+def _disable_auto_begin_commit(dbapi_connection):
+    # Disable underlying API's own emitting of BEGIN and COMMIT so we can ourselves
+    # https://docs.sqlalchemy.org/en/13/dialects/sqlite.html#serializable-isolation-savepoints-transactional-ddl
+    dbapi_connection.isolation_level = None
+
+
+def _is_sqlite_connection(dbapi_connection):
+    return isinstance(dbapi_connection, sqlite3.Connection)
+
+
+def _enable_sqlite_foreign_key_constraints(dbapi_connection):
+    cursor = dbapi_connection.cursor()
+    cursor.execute("PRAGMA foreign_keys=ON")
+    cursor.close()

src/cs50/_statement.py

@@ -0,0 +1,269 @@
+import collections
+import datetime
+import enum
+import re
+
+import sqlalchemy
+import sqlparse
+
+
+class Statement:
+    def __init__(self, dialect, sql, *args, **kwargs):
+        if len(args) > 0 and len(kwargs) > 0:
+            raise RuntimeError("cannot pass both positional and named parameters")
+
+        self._dialect = dialect
+        self._sql = sql
+        self._args = args
+        self._kwargs = kwargs
+
+        self._statement = self._parse()
+        self._command = self._parse_command()
+        self._tokens = self._bind_params()
+
+    def _parse(self):
+        formatted_statements = sqlparse.format(self._sql, strip_comments=True).strip()
+        parsed_statements = sqlparse.parse(formatted_statements)
+        num_of_statements = len(parsed_statements)
+        if num_of_statements == 0:
+            raise RuntimeError("missing statement")
+        elif num_of_statements > 1:
+            raise RuntimeError("too many statements at once")
+
+        return parsed_statements[0]
+
+
+    def _parse_command(self):
+        for token in self._statement:
+            if token.ttype in [sqlparse.tokens.Keyword, sqlparse.tokens.Keyword.DDL, sqlparse.tokens.Keyword.DML]:
+                token_value = token.value.upper()
+                if token_value in ["BEGIN", "DELETE", "INSERT", "SELECT", "START", "UPDATE"]:
+                    command = token_value
+                    break
+        else:
+            command = None
+
+        return command
+
+
+    def _bind_params(self):
+        tokens = self._tokenize()
+        paramstyle, placeholders = self._parse_placeholders(tokens)
+        if paramstyle in [Paramstyle.FORMAT, Paramstyle.QMARK]:
+            tokens = self._bind_format_or_qmark(placeholders, tokens)
+        elif paramstyle == Paramstyle.NUMERIC:
+            tokens = self._bind_numeric(placeholders, tokens)
+        if paramstyle in [Paramstyle.NAMED, Paramstyle.PYFORMAT]:
+            tokens = self._bind_named_or_pyformat(placeholders, tokens)
+
+        tokens = _escape_verbatim_colons(tokens)
+        return tokens
+
+
+    def _tokenize(self):
+        return list(self._statement.flatten())
+
+
+    def _parse_placeholders(self, tokens):
+        paramstyle = None
+        placeholders = collections.OrderedDict()
+        for index, token in enumerate(tokens):
+            if _is_placeholder(token):
+                _paramstyle, name = _parse_placeholder(token)
+                if paramstyle is None:
+                    paramstyle = _paramstyle
+                elif _paramstyle != paramstyle:
+                    raise RuntimeError("inconsistent paramstyle")
+
+                placeholders[index] = name
+
+        if paramstyle is None:
+            paramstyle = self._default_paramstyle()
+
+        return paramstyle, placeholders
+
+
+    def _default_paramstyle(self):
+        paramstyle = None
+        if self._args:
+            paramstyle = Paramstyle.QMARK
+        elif self._kwargs:
+            paramstyle = Paramstyle.NAMED
+
+        return paramstyle
+
+
+    def _bind_format_or_qmark(self, placeholders, tokens):
+        if len(placeholders) != len(self._args):
+            _placeholders = ", ".join([str(token) for token in placeholders.values()])
+            _args = ", ".join([str(self._escape(arg)) for arg in self._args])
+            if len(placeholders) < len(self._args):
+                raise RuntimeError(f"fewer placeholders ({_placeholders}) than values ({_args})")
+
+            raise RuntimeError(f"more placeholders ({_placeholders}) than values ({_args})")
+
+        for arg_index, token_index in enumerate(placeholders.keys()):
+            tokens[token_index] = self._escape(self._args[arg_index])
+
+        return tokens
+
+
+    def _bind_numeric(self, placeholders, tokens):
+        unused_arg_indices = set(range(len(self._args)))
+        for token_index, num in placeholders.items():
+            if num >= len(self._args):
+                raise RuntimeError(f"missing value for placeholder ({num + 1})")
+
+            tokens[token_index] = self._escape(self._args[num])
+            unused_arg_indices.remove(num)
+
+        if len(unused_arg_indices) > 0:
+            unused_args = ", ".join([str(self._escape(self._args[i])) for i in sorted(unused_arg_indices)])
+            raise RuntimeError(f"unused value{'' if len(unused_arg_indices) == 1 else 's'} ({unused_args})")
+
+        return tokens
+
+
+    def _bind_named_or_pyformat(self, placeholders, tokens):
+        unused_params = set(self._kwargs.keys())
+        for token_index, param_name in placeholders.items():
+            if param_name not in self._kwargs:
+                raise RuntimeError(f"missing value for placeholder ({param_name})")
+
+            tokens[token_index] = self._escape(self._kwargs[param_name])
+            unused_params.remove(param_name)
+
+        if len(unused_params) > 0:
+            raise RuntimeError("unused value{'' if len(unused_params) == 1 else 's'} ({', '.join(sorted(unused_params))})")
+
+        return tokens
+
+
+    def _escape(self, value):
+        """
+        Escapes value using engine's conversion function.
+        https://docs.sqlalchemy.org/en/latest/core/type_api.html#sqlalchemy.types.TypeEngine.literal_processor
+        """
+
+        if isinstance(value, (list, tuple)):
+            return self._escape_iterable(value)
+
+        if isinstance(value, bool):
+            return sqlparse.sql.Token(
+                sqlparse.tokens.Number,
+                sqlalchemy.types.Boolean().literal_processor(self._dialect)(value))
+
+        if isinstance(value, bytes):
+            if self._dialect.name in ["mysql", "sqlite"]:
+                # https://dev.mysql.com/doc/refman/8.0/en/hexadecimal-literals.html
+                return sqlparse.sql.Token(sqlparse.tokens.Other, f"x'{value.hex()}'")
+            if self._dialect.name in ["postgres", "postgresql"]:
+                # https://dba.stackexchange.com/a/203359
+                return sqlparse.sql.Token(sqlparse.tokens.Other, f"'\\x{value.hex()}'")
+
+            raise RuntimeError(f"unsupported value: {value}")
+
+        if isinstance(value, datetime.date):
+            return sqlparse.sql.Token(
+                sqlparse.tokens.String,
+                sqlalchemy.types.String().literal_processor(self._dialect)(value.strftime("%Y-%m-%d")))
+
+        if isinstance(value, datetime.datetime):
+            return sqlparse.sql.Token(
+                sqlparse.tokens.String,
+                sqlalchemy.types.String().literal_processor(self._dialect)(value.strftime("%Y-%m-%d %H:%M:%S")))
+
+        if isinstance(value, datetime.time):
+            return sqlparse.sql.Token(
+                sqlparse.tokens.String,
+                sqlalchemy.types.String().literal_processor(self._dialect)(value.strftime("%H:%M:%S")))
+
+        if isinstance(value, float):
+            return sqlparse.sql.Token(
+                sqlparse.tokens.Number,
+                sqlalchemy.types.Float().literal_processor(self._dialect)(value))
+
+        if isinstance(value, int):
+            return sqlparse.sql.Token(
+                sqlparse.tokens.Number,
+                sqlalchemy.types.Integer().literal_processor(self._dialect)(value))
+
+        if isinstance(value, str):
+            return sqlparse.sql.Token(
+                sqlparse.tokens.String,
+                sqlalchemy.types.String().literal_processor(self._dialect)(value))
+
+        if value is None:
+            return sqlparse.sql.Token(
+                sqlparse.tokens.Keyword,
+                sqlalchemy.types.NullType().literal_processor(self._dialect)(value))
+
+        raise RuntimeError(f"unsupported value: {value}")
+
+
+    def _escape_iterable(self, iterable):
+        return sqlparse.sql.TokenList(
+            sqlparse.parse(", ".join([str(self._escape(v)) for v in iterable])))
+
+
+    def get_command(self):
+        return self._command
+
+
+    def __str__(self):
+        return "".join([str(token) for token in self._tokens])
+
+
+def _is_placeholder(token):
+    return token.ttype == sqlparse.tokens.Name.Placeholder
+
+
+def _parse_placeholder(token):
+    if token.value == "?":
+        return Paramstyle.QMARK, None
+
+    # E.g., :1
+    matches = re.search(r"^:([1-9]\d*)$", token.value)
+    if matches:
+        return Paramstyle.NUMERIC, int(matches.group(1)) - 1
+
+    # E.g., :foo
+    matches = re.search(r"^:([a-zA-Z]\w*)$", token.value)
+    if matches:
+        return Paramstyle.NAMED, matches.group(1)
+
+    if token.value == "%s":
+        return Paramstyle.FORMAT, None
+
+    # E.g., %(foo)
+    matches = re.search(r"%\((\w+)\)s$", token.value)
+    if matches:
+        return Paramstyle.PYFORMAT, matches.group(1)
+
+    raise RuntimeError(f"{token.value}: invalid placeholder")
+
+
+def _escape_verbatim_colons(tokens):
+    for token in tokens:
+        if _is_string_literal(token):
+            token.value = re.sub("(^'|\s+):", r"\1\:", token.value)
+        elif _is_identifier(token):
+            token.value = re.sub("(^\"|\s+):", r"\1\:", token.value)
+
+    return tokens
+
+
+def _is_string_literal(token):
+    return token.ttype in [sqlparse.tokens.Literal.String, sqlparse.tokens.Literal.String.Single]
+
+
+def _is_identifier(token):
+    return token.ttype == sqlparse.tokens.Literal.String.Symbol
+
+
+class Paramstyle(enum.Enum):
+    FORMAT = enum.auto()
+    NAMED = enum.auto()
+    NUMERIC = enum.auto()
+    PYFORMAT = enum.auto()
+    QMARK = enum.auto()

src/cs50/cs50.py

@@ -1,98 +1,6 @@
-from __future__ import print_function
-
-import inspect
-import logging
-import os
 import re
 import sys
 
-from distutils.sysconfig import get_python_lib
-from os.path import abspath, join
-from termcolor import colored
-from traceback import format_exception
-
-
-# Configure default logging handler and formatter
-# Prevent flask, werkzeug, etc from adding default handler
-logging.basicConfig(format="%(levelname)s: %(message)s", level=logging.DEBUG)
-
-try:
-    # Patch formatException
-    logging.root.handlers[0].formatter.formatException = lambda exc_info: _formatException(*exc_info)
-except IndexError:
-    pass
-
-# Configure cs50 logger
-_logger = logging.getLogger("cs50")
-_logger.setLevel(logging.DEBUG)
-
-# Log messages once
-_logger.propagate = False
-
-handler = logging.StreamHandler()
-handler.setLevel(logging.DEBUG)
-
-formatter = logging.Formatter("%(levelname)s: %(message)s")
-formatter.formatException = lambda exc_info: _formatException(*exc_info)
-handler.setFormatter(formatter)
-_logger.addHandler(handler)
-
-
-class _flushfile():
-    """
-    Disable buffering for standard output and standard error.
-
-    http://stackoverflow.com/a/231216
-    """
-
-    def __init__(self, f):
-        self.f = f
-
-    def __getattr__(self, name):
-        return object.__getattribute__(self.f, name)
-
-    def write(self, x):
-        self.f.write(x)
-        self.f.flush()
-
-
-sys.stderr = _flushfile(sys.stderr)
-sys.stdout = _flushfile(sys.stdout)
-
-
-def _formatException(type, value, tb):
-    """
-    Format traceback, darkening entries from global site-packages directories
-    and user-specific site-packages directory.
-
-    https://stackoverflow.com/a/46071447/5156190
-    """
-
-    # Absolute paths to site-packages
-    packages = tuple(join(abspath(p), "") for p in sys.path[1:])
-
-    # Highlight lines not referring to files in site-packages
-    lines = []
-    for line in format_exception(type, value, tb):
-        matches = re.search(r"^  File \"([^\"]+)\", line \d+, in .+", line)
-        if matches and matches.group(1).startswith(packages):
-            lines += line
-        else:
-            matches = re.search(r"^(\s*)(.*?)(\s*)$", line, re.DOTALL)
-            lines.append(matches.group(1) + colored(matches.group(2), "yellow") + matches.group(3))
-    return "".join(lines).rstrip()
-
-
-sys.excepthook = lambda type, value, tb: print(_formatException(type, value, tb), file=sys.stderr)
-
-
-def eprint(*args, **kwargs):
-    raise RuntimeError("The CS50 Library for Python no longer supports eprint, but you can use print instead!")
-
-
-def get_char(prompt):
-    raise RuntimeError("The CS50 Library for Python no longer supports get_char, but you can use get_string instead!")
-
 
 def get_float(prompt):
     """
@@ -101,14 +9,21 @@ def get_float(prompt):
     prompted to retry. If line can't be read, return None.
     """
     while True:
-        s = get_string(prompt)
-        if s is None:
-            return None
-        if len(s) > 0 and re.search(r"^[+-]?\d*(?:\.\d*)?$", s):
-            try:
-                return float(s)
-            except (OverflowError, ValueError):
-                pass
+        try:
+            return _get_float(prompt)
+        except (OverflowError, ValueError):
+            pass
+
+
+def _get_float(prompt):
+    s = get_string(prompt)
+    if s is None:
+        return
+
+    if len(s) > 0 and re.search(r"^[+-]?\d*(?:\.\d*)?$", s):
+        return float(s)
+
+    raise ValueError(f"invalid float literal: {s}")
 
 
 def get_int(prompt):
@@ -118,14 +33,21 @@ def get_int(prompt):
     can't be read, return None.
     """
     while True:
-        s = get_string(prompt)
-        if s is None:
-            return None
-        if re.search(r"^[+-]?\d+$", s):
-            try:
-                return int(s, 10)
-            except ValueError:
-                pass
+        try:
+            return  _get_int(prompt)
+        except (MemoryError, ValueError):
+            pass
+
+
+def _get_int(prompt):
+    s = get_string(prompt)
+    if s is None:
+        return
+
+    if re.search(r"^[+-]?\d+$", s):
+        return int(s, 10)
+
+    raise ValueError(f"invalid int literal for base 10: {s}")
 
 
 def get_string(prompt):
@@ -137,7 +59,35 @@ def get_string(prompt):
     """
     if type(prompt) is not str:
         raise TypeError("prompt must be of type str")
+
     try:
-        return input(prompt)
+        return _get_input(prompt)
     except EOFError:
-        return None
+        return
+
+
+def _get_input(prompt):
+    return input(prompt)
+
+
+class _flushfile():
+    """
+    Disable buffering for standard output and standard error.
+    http://stackoverflow.com/a/231216
+    """
+
+    def __init__(self, f):
+        self.f = f
+
+    def __getattr__(self, name):
+        return object.__getattribute__(self.f, name)
+
+    def write(self, x):
+        self.f.write(x)
+        self.f.flush()
+
+def disable_buffering():
+    sys.stderr = _flushfile(sys.stderr)
+    sys.stdout = _flushfile(sys.stdout)
+
+disable_buffering()

src/cs50/sql.py

@@ -0,0 +1,151 @@
+import math
+import sys
+import unittest
+
+from unittest.mock import patch
+
+from cs50.cs50 import get_string, _get_int, _get_float
+
+
+class TestCS50(unittest.TestCase):
+    @patch("cs50.cs50._get_input", return_value="")
+    def test_get_string_empty_input(self, mock_get_input):
+        """Returns empty string when input is empty"""
+        self.assertEqual(get_string("Answer: "), "")
+        mock_get_input.assert_called_with("Answer: ")
+
+
+    @patch("cs50.cs50._get_input", return_value="test")
+    def test_get_string_nonempty_input(self, mock_get_input):
+        """Returns the provided non-empty input"""
+        self.assertEqual(get_string("Answer: "), "test")
+        mock_get_input.assert_called_with("Answer: ")
+
+
+    @patch("cs50.cs50._get_input", side_effect=EOFError)
+    def test_get_string_eof(self, mock_get_input):
+        """Returns None on EOF"""
+        self.assertIs(get_string("Answer: "), None)
+        mock_get_input.assert_called_with("Answer: ")
+
+
+    def test_get_string_invalid_prompt(self):
+        """Raises TypeError when prompt is not str"""
+        with self.assertRaises(TypeError):
+            get_string(1)
+
+
+    @patch("cs50.cs50.get_string", return_value=None)
+    def test_get_int_eof(self, mock_get_string):
+        """Returns None on EOF"""
+        self.assertIs(_get_int("Answer: "), None)
+        mock_get_string.assert_called_with("Answer: ")
+
+
+    def test_get_int_valid_input(self):
+        """Returns the provided integer input"""
+
+        def assert_equal(return_value, expected_value):
+            with patch("cs50.cs50.get_string", return_value=return_value) as mock_get_string:
+                self.assertEqual(_get_int("Answer: "), expected_value)
+                mock_get_string.assert_called_with("Answer: ")
+
+        values = [
+            ("0", 0),
+            ("50", 50),
+            ("+50", 50),
+            ("+42", 42),
+            ("-42", -42),
+            ("42", 42),
+        ]
+
+        for return_value, expected_value in values:
+            assert_equal(return_value, expected_value)
+
+
+    def test_get_int_invalid_input(self):
+        """Raises ValueError when input is invalid base-10 int"""
+
+        def assert_raises_valueerror(return_value):
+            with patch("cs50.cs50.get_string", return_value=return_value) as mock_get_string:
+                with self.assertRaises(ValueError):
+                    _get_int("Answer: ")
+
+                mock_get_string.assert_called_with("Answer: ")
+
+        return_values = [
+            "++50",
+            "--50",
+            "50+",
+            "50-",
+            " 50",
+            " +50",
+            " -50",
+            "50 ",
+            "ab50",
+            "50ab",
+            "ab50ab",
+        ]
+
+        for return_value in return_values:
+            assert_raises_valueerror(return_value)
+
+
+    @patch("cs50.cs50.get_string", return_value=None)
+    def test_get_float_eof(self, mock_get_string):
+        """Returns None on EOF"""
+        self.assertIs(_get_float("Answer: "), None)
+        mock_get_string.assert_called_with("Answer: ")
+
+
+    def test_get_float_valid_input(self):
+        """Returns the provided integer input"""
+        def assert_equal(return_value, expected_value):
+            with patch("cs50.cs50.get_string", return_value=return_value) as mock_get_string:
+                f = _get_float("Answer: ")
+                self.assertTrue(math.isclose(f, expected_value))
+                mock_get_string.assert_called_with("Answer: ")
+
+        values = [
+            (".0", 0.0),
+            ("0.", 0.0),
+            (".42", 0.42),
+            ("42.", 42.0),
+            ("50", 50.0),
+            ("+50", 50.0),
+            ("-50", -50.0),
+            ("+3.14", 3.14),
+            ("-3.14", -3.14),
+        ]
+
+        for return_value, expected_value in values:
+            assert_equal(return_value, expected_value)
+
+
+    def test_get_float_invalid_input(self):
+        """Raises ValueError when input is invalid float"""
+
+        def assert_raises_valueerror(return_value):
+            with patch("cs50.cs50.get_string", return_value=return_value) as mock_get_string:
+                with self.assertRaises(ValueError):
+                    _get_float("Answer: ")
+
+                mock_get_string.assert_called_with("Answer: ")
+
+        return_values = [
+            ".",
+            "..5",
+            "a.5",
+            ".5a"
+            "0.5a",
+            "a0.42",
+            " .42",
+            "3.14 ",
+            "++3.14",
+            "3.14+",
+            "--3.14",
+            "3.14--",
+        ]
+
+        for return_value in return_values:
+            assert_raises_valueerror(return_value)