Develop/main (#14)

migrate to **v2.1.0**:
* update response schema
* use only debug-level logging
* rearrange and simplify code

Author: hell03end

.gitignore

@@ -1,3 +1,4 @@
+.pytest_cache/
 __pycache__/
 .vscode/
 .cache/

requirements.txt

@@ -1,3 +1,3 @@
+# dev dependencies
 pytest>=3.2.1
-requests==2.11.1
 twine==1.9.1

ruz/__init__.py

@@ -1,17 +1,16 @@
 """
-    Python wrapper for HSE RUZ API
+    Python wrapper for HSE RUZ API.
 
     Usage
     -----
     import ruz
     assert ruz.person_lessons("mymail@edu.hse.ru")
 """
 
-from ruz.main import (auditoriums, buildings, chairs, faculties, find_by_str,
-                      get_formated_date, groups, is_hse_email, is_student,
-                      is_valid_hse_email, kind_of_works, lecturers,
-                      person_lessons, schedules, staff_of_group, streams,
-                      sub_groups, type_of_auditoriums)
+from ruz.api import (auditoriums, buildings, chairs, faculties, find_by_str,
+                     groups, kind_of_works, lecturers, person_lessons,
+                     schedules, staff_of_group, streams, sub_groups,
+                     type_of_auditoriums)
 
 __author__ = "Dmitriy Pchelkin | hell03end"
-__version__ = (2, 0, 1)
+__version__ = (2, 1, 0)

ruz/main.py ruz/api.pyruz/main.py renamed to ruz/api.py

@@ -1,189 +1,10 @@
-import json
-import re
+import logging
 from collections import Callable, Iterable
-from datetime import datetime, timedelta
 from functools import lru_cache
-from urllib import error, parse, request
 
-from ruz.logging import log, logging
-from ruz.schema import REQUEST_SCHEMA, RUZ_API_ENDPOINTS
-from ruz.utils import (CHECK_EMAIL_ONLINE, RUZ_API_URL, RUZ_API_V,
-                       USE_NONE_SAFE_VALUES, none_safe)
+from ruz.utils import get, get_formated_date, is_student
 
 
-# ===== Common methods =====
-
-def is_student(email: str) -> bool or None:
-    """
-        Check email belongs to student
-
-        :param email, required - valid HSE email addres or domain.
-
-        Stutent's domain:   @edu.hse.ru
-        HSE stuff' domain:  @hse.ru
-    """
-    email_domain = email.lower().split("@")[-1]
-
-    if email_domain == "edu.hse.ru":
-        return True
-    elif email_domain == "hse.ru":
-        return False
-
-    logging.error("Wrong HSE email domain: '%s'", email_domain)
-
-
-def is_hse_email(email: str) -> bool:
-    """
-        Check email is valid HSE corp. email
-
-        :param email, required - email address to check.
-    """
-    if re.fullmatch(r"^[a-z0-9\._-]{3,}@(edu\.)?hse\.ru$", email.lower()):
-        return True
-    logging.debug("Incorrect HSE email '%s'.", email)
-    return False
-
-
-def get_formated_date(day_bias: int or float=0) -> str:
-    """
-        Return date in RUZ API compatible format
-
-        :param day_bias - number of day from now.
-    """
-    return (datetime.now() + timedelta(
-        days=float(day_bias)
-    )).strftime("%Y.%m.%d")
-
-
-@log()
-def is_valid_hse_email(email: str) -> bool:
-    """
-        Check email is valid via API endpoint call (schedule)
-
-        :param email - email address to check (for schedules only).
-    """
-    @none_safe()
-    def request_schedule_api(**params) -> list or dict:
-        return request.urlopen(make_url(
-            "schedule",
-            email=email,
-            fromDate=get_formated_date(),
-            toDate=get_formated_date(1),
-            **params
-        ))
-
-    email = email.strip().lower()
-    if not is_hse_email(email):
-        return False
-
-    try:
-        response = request_schedule_api(
-            receiverType=1 if not is_student(email) else None
-        )
-        del response
-    except (error.HTTPError, error.URLError) as err:
-        logging.debug("Email '%s' wasn't verified.\n%s", email, err)
-        return False
-    return True
-
-
-# ===== Special methods =====
-
-@log()
-def is_valid_schema(endpoint: str,
-                    check_email_online: bool=CHECK_EMAIL_ONLINE,
-                    **params) -> bool:
-    """
-        Check params fit schema for certain endpoint
-
-        :param endpoint - endpoint for request.
-        :param check_email_online - use is_valid_hse_email.
-        :param params - schema params.
-    """
-
-    if (endpoint == "schedule" and "lecturerOid" not in params and
-            "studentOid" not in params and "email" not in params and
-            "auditoriumOid" not in params):
-        logging.debug("One of the followed required: lecturer_id, "
-                      "auditorium_id, student_id, email for "
-                      "schedule endpoint.")
-        return False
-
-    if params.get('email') is not None:
-        email = params['email']
-        if not is_hse_email(email):
-            del email
-            return False
-        elif check_email_online and not is_valid_hse_email(email):
-            logging.warning("'%s' is not verified by API call.", email)
-        del email
-
-    endpoint = RUZ_API_ENDPOINTS.get(endpoint)
-    if endpoint is None:
-        logging.warning("Can't find endpoint: '%s'.", endpoint)
-        del endpoint
-        return False
-
-    schema = REQUEST_SCHEMA[endpoint]
-    for key, value in params.items():
-        if key not in schema:
-            logging.warning("Can't find '%s' schema param: '%s'",
-                            endpoint, key)
-            del schema, endpoint
-            return False
-        if not isinstance(value, schema[key]):
-            logging.warning("Expected {} for '{}'::'{}' got: {}",
-                            schema[key], endpoint, key, type(value))
-            del schema, endpoint
-            return False
-    del schema, endpoint
-    return True
-
-
-@log()
-def make_url(endpoint: str, **params) -> str:
-    """
-        Creates URL for API requests
-
-        :param endpoint - endpoint for request.
-        :param params - request params.
-    """
-    url = "".join((RUZ_API_URL, RUZ_API_ENDPOINTS[endpoint]))
-    if params:
-        return "?".join((url, parse.urlencode(params)))
-    return url
-
-
-@none_safe()
-@log()
-def get(endpoint: str,
-        encoding: str="utf-8",
-        return_none_safe: bool=USE_NONE_SAFE_VALUES,
-        **params) -> (list, dict, None):
-    """
-        Return requested data in JSON
-
-        Check request has correct schema.
-
-        :param endpoint - endpoint for request.
-        :param encoding - encoding for received data.
-        :param return_none_safe - return empty list on fallback.
-        :param params - requested params
-    """
-    if not is_valid_schema(endpoint, **params):
-        return [] if return_none_safe else None
-
-    url = make_url(endpoint, **params)
-    try:
-        response = request.urlopen(url)
-        return json.loads(response.read().decode(encoding))
-    except (error.HTTPError, error.URLError) as err:
-        logging.warning("Can't get '%s'.\n%s", url, err)
-    return [] if return_none_safe else None
-
-
-# ===== API methods =====
-
 def schedules(emails: Iterable=None,
               lecturer_ids: Iterable=None,
               auditorium_ids: Iterable=None,
@@ -385,8 +206,6 @@ def sub_groups(reset_cache: bool=False) -> list:
     return get("subGroups")
 
 
-# ===== Additional methods =====
-
 def find_by_str(subject: str or Callable,
                 query: str,
                 by: str="name",
@@ -440,4 +259,5 @@ def find_by_str(subject: str or Callable,
         raise NotImplementedError(subject.__name__)
 
     query = query.strip().lower()
-    return [el for el in subject(**params) if query in el[by].lower().strip()]
+    return [el for el in subject(**params)
+            if query in (el[by].lower().strip() if el[by] else "")]

ruz/logging.py

@@ -1,121 +1,24 @@
 import logging
-import traceback
+import os
 from collections import Callable
 from functools import wraps
-from logging import RootLogger
 
-
-# Default logging behavior
-logging.basicConfig(
-    level=logging.WARNING,
-    format="[%(asctime)s] %(levelname)s "
-           "[%(name)s.{%(filename)s}.%(funcName)s:%(lineno)d] %(message)s",
-    datefmt="%H:%M:%S"
-)
-
-
-class Log:
-    """ Context manager for events' logging. Handling (log) exceptions. """
-
-    def __init__(self,
-                 case_name: str,
-                 level: int=logging.DEBUG,
-                 logger: RootLogger=None,
-                 **kwargs) -> None:
-        self._name = case_name
-        self._level = level
-        self._logger = logger if logger is not None else None
-        self._enter_msg = kwargs.pop("enter_msg", "ENTERING::")
-        self._exit_msg = kwargs.pop("exit_msg", "EXITING::")
-        self._exc_msg = kwargs.pop("exc_msg", "")
-        self._silent = kwargs.pop("silent", False)
-
-    def _log(self, *message, level: int=None) -> None:
-        """ log message """
-        if level is None:
-            level = self._level
-
-        if level == logging.ERROR:
-            if self._logger:
-                self._logger.error(*message)
-            else:
-                logging.error(*message)
-        elif level > logging.DEBUG:
-            if self._logger:
-                self._logger.info(*message)
-            else:
-                logging.info(*message)
-        else:
-            if self._logger:
-                self._logger.debug(*message)
-            else:
-                logging.debug(*message)
-
-    def __enter__(self) -> object:
-        """ Returns it's logger instance of self if silent """
-        if not self._silent:
-            self._log("%s%s", self._enter_msg, self._name)
-        return self
-
-    def __exit__(self,
-                 exc_type: object=None,
-                 exc_val: object=None,
-                 tb: object=None) -> None:
-        """ Handling (log) exceptions """
-        if exc_type is not None:
-            self._log(
-                "%s\n%s%s: %s\n",
-                self._exc_msg,
-                "\n".join([s.strip(r"\n") for s in traceback.format_tb(tb)]),
-                exc_type.__name__,
-                exc_val
-            )
-
-        if not self._silent:
-            self._log("%s%s", self._exit_msg, self._name)
-
-
-class _FuncLog(Log):
-    """ Context manager for logging function calls """
-
-    def __init__(self,
-                 case_name: str,
-                 level: int=logging.DEBUG,
-                 logger: RootLogger=None):
-        super(_FuncLog, self).__init__(
-            case_name=case_name,
-            level=level,
-            enter_msg="===> ",
-            exit_msg="<--- ",
-            logger=logger
-        )
-
-
-def log(log_result: bool=True,
-        log_args: bool=True,
-        log_kwargs: bool=True,
-        name: str=None,
-        level: int=logging.DEBUG,
-        logger: RootLogger=None) -> Callable:
-    """ Returns decorator for logging function/method behavior """
-    def decor(func: Callable) -> Callable:
-        func_name = name
-        if not func_name:
-            func_name = func.__name__
-
-        @wraps(func)
-        def wrapper(*args, **kwargs) -> object:
-            with _FuncLog(func_name, level, logger=logger):
-                if log_args:
-                    for arg in args:
-                        logging.debug("()::%s", arg)
-                if log_kwargs:
-                    for key, value in kwargs.items():
-                        logging.debug(r"{}::%s=%s", key, value)
-
-                result = func(*args, **kwargs)
-                if log_result:
-                    logging.debug("RETURN(%s)::%s", func_name, result)
-            return result
-        return wrapper
-    return decor
+ENABLE_LOGGING = os.environ.get("HSE_RUZ_ENABLE_LOGGING", True)
+
+
+def log(func: Callable) -> Callable:
+    if not ENABLE_LOGGING:
+        return func
+
+    @wraps(func)
+    def wrapper(*args, **kwargs) -> object:
+        func_name = func.__name__
+        logging.debug("[%s]\tENTER", func_name)
+        for arg in args:
+            logging.debug("[%s]\tARG\t%s", func_name, arg)
+        for key, value in kwargs.items():
+            logging.debug("[%s]\tKWARG\t%s=%s", func_name, key, value)
+        result = func(*args, **kwargs)
+        logging.debug("[%s]\tEXIT", func_name)
+        return result
+    return wrapper