From 9b8820e43d7c164f7cef3372224d6b9d6975ecef Mon Sep 17 00:00:00 2001 From: Miguel Barão Date: Tue, 21 Feb 2023 20:00:39 +0000 Subject: [PATCH] move run_script to questions.py --- aprendizations/questions.py | 52 +++++++++++++++++++++++++++++++++++++++++++++++----- aprendizations/tools.py | 78 +++++++++++++++++++++++++++++++++++++++--------------------------------------- 2 files changed, 86 insertions(+), 44 deletions(-) diff --git a/aprendizations/questions.py b/aprendizations/questions.py index 89c7ced..24bf10f 100644 --- a/aprendizations/questions.py +++ b/aprendizations/questions.py @@ -11,11 +11,11 @@ import logging from os import path import random import re -from typing import Any, Dict, NewType +from typing import Any, Dict, List, NewType import uuid -# this project -from aprendizations.tools import run_script +# third party libraries +import yaml # setup logger for this module logger = logging.getLogger(__name__) @@ -33,8 +33,9 @@ class QuestionException(Exception): # ============================================================================ class Question(dict): ''' - Classes derived from this base class are meant to instantiate questions - for each student. + Base class for all question types + + Classes derived from this base class can instantiate questions. Instances can shuffle options or automatically generate questions. ''' @@ -695,3 +696,44 @@ class QFactory(): question = question_from(qdict) # returns a Question instance question.gen() return question + +# ----------------------------------------------------------------------------- +async def run_script(script: str, + args: List[str], + stdin: str = '', + timeout: int = 5) -> Any: + + # normalize args + script = path.expanduser(script) + input_bytes = stdin.encode('utf-8') + args = [str(a) for a in args] + + try: + p = await asyncio.create_subprocess_exec( + script, *args, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + ) + except FileNotFoundError: + logger.error(f'Can not execute script "{script}": not found.') + except PermissionError: + logger.error(f'Can not execute script "{script}": wrong permissions.') + except OSError: + logger.error(f'Can not execute script "{script}": unknown reason.') + else: + try: + stdout, _ = await asyncio.wait_for(p.communicate(input_bytes), timeout) + except asyncio.TimeoutError: + logger.warning(f'Timeout {timeout}s exceeded running "{script}".') + return + + if p.returncode != 0: + logger.error(f'Return code {p.returncode} running "{script}".') + else: + try: + output = yaml.safe_load(stdout.decode('utf-8', 'ignore')) + except yaml.YAMLError: + logger.error(f'Error parsing yaml output of "{script}"') + else: + return output diff --git a/aprendizations/tools.py b/aprendizations/tools.py index 75ad213..e53b005 100644 --- a/aprendizations/tools.py +++ b/aprendizations/tools.py @@ -212,42 +212,42 @@ def load_yaml(filename: str, default: Any = None) -> Any: # ---------------------------------------------------------------------------- # Same as above, but asynchronous # ---------------------------------------------------------------------------- -async def run_script(script: str, - args: List[str] = [], - stdin: str = '', - timeout: int = 2) -> Any: - - # normalize args - script = path.expanduser(script) - input_bytes = stdin.encode('utf-8') - args = [str(a) for a in args] - - try: - p = await asyncio.create_subprocess_exec( - script, *args, - stdin=asyncio.subprocess.PIPE, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.DEVNULL, - ) - except FileNotFoundError: - logger.error(f'Can not execute script "{script}": not found.') - except PermissionError: - logger.error(f'Can not execute script "{script}": wrong permissions.') - except OSError: - logger.error(f'Can not execute script "{script}": unknown reason.') - else: - try: - stdout, _ = await asyncio.wait_for(p.communicate(input_bytes), timeout) - except asyncio.TimeoutError: - logger.warning(f'Timeout {timeout}s exceeded running "{script}".') - return - - if p.returncode != 0: - logger.error(f'Return code {p.returncode} running "{script}".') - else: - try: - output = yaml.safe_load(stdout.decode('utf-8', 'ignore')) - except Exception: - logger.error(f'Error parsing yaml output of "{script}"') - else: - return output +# async def run_script(script: str, +# args: List[str] = [], +# stdin: str = '', +# timeout: int = 5) -> Any: +# +# # normalize args +# script = path.expanduser(script) +# input_bytes = stdin.encode('utf-8') +# args = [str(a) for a in args] +# +# try: +# p = await asyncio.create_subprocess_exec( +# script, *args, +# stdin=asyncio.subprocess.PIPE, +# stdout=asyncio.subprocess.PIPE, +# stderr=asyncio.subprocess.DEVNULL, +# ) +# except FileNotFoundError: +# logger.error(f'Can not execute script "{script}": not found.') +# except PermissionError: +# logger.error(f'Can not execute script "{script}": wrong permissions.') +# except OSError: +# logger.error(f'Can not execute script "{script}": unknown reason.') +# else: +# try: +# stdout, _ = await asyncio.wait_for(p.communicate(input_bytes), timeout) +# except asyncio.TimeoutError: +# logger.warning(f'Timeout {timeout}s exceeded running "{script}".') +# return +# +# if p.returncode != 0: +# logger.error(f'Return code {p.returncode} running "{script}".') +# else: +# try: +# output = yaml.safe_load(stdout.decode('utf-8', 'ignore')) +# except Exception: +# logger.error(f'Error parsing yaml output of "{script}"') +# else: +# return output -- libgit2 0.21.2