4
\$\begingroup\$

I wrote the Validator class. In the project, the presented code is located at the path ".../validator/init.py". It is needed for testing code in different virtual Conda environments, in the virtual environments PyTorch with the necessary versions are prepared in advance. The code works properly. But, I want to improve its reliability, readability and extensibility.

The results of the function run_test_old_and_new_code I will apply for calculations of RAG efficiency. At the moment, the metric is not complicated. Code works - True, does not work - False. I create temporary files for executing the code and run them using subprocess.run(). Also, I added logging.

import logging
import time
LOGGER = logging.getLogger(__name__)
start = time.time()
LOGGER.debug(f"start init library")

import os 
import logging
import subprocess
import tempfile

# Patch for conda.exe 
CONDA_PATH = 'C:/Miniconda3/Scripts/conda.exe'
if not os.path.exists(CONDA_PATH):
    LOGGER.error(f"Conda {CONDA_PATH} .")
    exit

LOGGER.debug(f"time init library: {time.time() - start} sec")


class Validator():

    def _dynamic_val(self, code, torch_version: str ='2.7.0'):
        with tempfile.NamedTemporaryFile(mode='w', suffix='.py', 
                                         delete=False, # NO AUTO DELETE !!
                                         encoding='utf-8') as temp_file:
            temp_file.write(code)
            temp_path = temp_file.name

        try:
            LOGGER.debug(f"Start validation")
            conda_env_ending = self._version_in_conda_env_ending(torch_version)
            LOGGER.debug(f"In env: {self._get_version_torch(conda_env_ending)}")

            runner_config = [CONDA_PATH, 'run', '-n', f'torch{conda_env_ending}', 'python', temp_path]
            result_sub = subprocess.run(runner_config, capture_output=True,
                                        encoding='utf-8')

            out = result_sub.stdout.strip()
            err = result_sub.stderr.strip()

            if result_sub.returncode != 0:
                LOGGER.error(f"ERROR in conda: {err}")
            else:
                LOGGER.debug(f'ERROR in code: {temp_path})')
            return out, err
        except Exception as e:
            LOGGER.error(f"ERROR in validation: {e}")
        


    def _get_version_torch(self, conda_env_ending):
        try:
            version_cmd = [
                CONDA_PATH, 'run', '-n', f"torch{conda_env_ending}", 'python', '-c',
                'import torch; print(f"PyTorch version: {torch.__version__}")'
            ]
            out = subprocess.run(version_cmd, capture_output=True,
                                            encoding="utf-8").stdout.strip()
            return out
        except Exception as e: 
            LOGGER.error(f"_get_version_torch ERROR: {e}")
            raise

    def _version_in_conda_env_ending(self, version:str)->str:
        v_list = version.split('.')
        return ''.join(v_list)
    
    def run_test_code(self, code: str, env_version: str):
        '''
            Check works code on version env_version
        '''
        out, err = self._dynamic_val(code, torch_version=env_version)
        works = (err == '')
        return works, out, err
    

    def run_test_old_and_new_code(self, name_test, old_code, new_code):
        """
            Testing old and new code
        """
        LOGGER.info(f"[START TEST] {name_test}")
        
        works_100, out_100, err_100 = self.run_test_code(old_code, '1.0.0')
        LOGGER.info(f"{name_test} в 1.0.0: {'Work' if works_100 else 'Not work'}")
        LOGGER.debug(f"{name_test} в 1.0.0: {'Work' if works_100 else 'Not work'} | Out: {out_100} | Err: {err_100}")
        
        # Test 2.7.0
        works_270, out_270, err_270 = self.run_test_code(old_code, '2.7.0')
        LOGGER.info(f"{name_test} в 2.7.0: {'Work' if works_270 else 'Not work'}")
        LOGGER.debug(f"{name_test} в 2.7.0: {'Work' if works_270 else 'Not work'} | Out: {out_270} | Err: {err_270}")
        
        # Test answer LLM
        new_code_works, new_code_out, new_code_err = self.run_test_code(new_code, '2.7.0')
        LOGGER.info(f"Answer LLM RAG в 2.7.0: {'Work' if new_code_works else 'Not work'}")
        LOGGER.debug(f"Answer LLM RAG в 2.7.0: {'Work' if new_code_works else 'Not work'} | Out: {new_code_out} | Err: {new_code_err}")
        
        LOGGER.info(f"[END TEST] {name_test}")

        return {
            'API': name_test,
            'Is work in 1.0.0': works_100,
            'Out 1.0.0': out_100,
            'Err 1.0.0': err_100[:200], # Think need record errors some other way. In future will change.
            'Is work in 2.7.0': works_270,
            'Out 2.7.0': out_270,
            'Err 2.7.0': err_270[:200], # Think need record errors some other way. In future will change.
            'Is work in 2.7.0 (после RAG)': new_code_works, 
            'RAG Out 2.7.0': new_code_out,
            'RAG Err 2.7.0': new_code_err[:200], # Think need record errors some other way. In future will change.
        }, {
        'old_code': old_code,
            'new_code': new_code
        }

Am I engaging in the correct practice for working with conda and subprocess? Also, there is a feeling that I missed something in the error handlers.

In the logger settings, it is specified that the debug level is saved to a file. The info level is output to the console.

\$\endgroup\$

2 Answers 2

Reset to default
5
\$\begingroup\$

version hardcodes

To dry up run_test_old_and_new_code(), it looks like we want a helper that accepts a version number such as "1.0.0" or "2.7.0". In _dynamic_val() the 2.7.0 default is very nice, but it didn't carry over to this function. And 2.7.0 appears in many logging statements and calls. At some point 2.7.1 will be relevant, and we risk having a maintainer replace just a subset of the versions that will need updating.

error handling

Some of these try blocks seem to correspond to "impossible" or "never happens" conditions, which your automated tests do not try to exercise. If so, then consider removing the try, making the error fatal. The default stack trace display will be more informative than what the OP code displays. If there's some specific logging requirement in your specifications document, then deal with it at top level in a uniform way, instead of down within the various helpers.

OOP

There is no __init__() constructor, indeed, no self.foo attributes at all. This suggests that what you really wanted was a collection of functions, rather than methods hanging off an object instance.

It is usual to tack on a @staticmethod decorator on methods which don't rely on self. Here, we might have noticed that all the helpers are static, and the pair of entry points in the Public API only use self for dispatch to a helper.

temp file cleanup

There's no """docstring""" on this helper to explain the concept of operations. Consider having _dynamic_val() return a Path, which caller is expected to .unlink(), perhaps via a with context handler. Currently it appears that such files accumulate without bound.

wrong exit

if not os.path.exists(CONDA_PATH):
    ...
    exit

The two development process difficulties here are:

  1. This doesn't work.
  2. You never ran that line of code.

The first is easy: add import sys and call sys.exit()

The second is tougher, since it corresponds to adopting a brand new habit and exercising it each day that you write code. I have personally written error handlers like this, which error at exactly the time you were hoping they would instead be helpful. Recommend that you introduce a deliberate typographic error ("C:/does/not/exist") when defining that path, and verify it's handled as you desire. Then revert the typo, and don't worry about that clause until the next time you mess with the if test or the body.

A pretty common way to lose in such a handler is with NameError.

# foo = 42
if is_failed():
    print("failed, with foo =", foo)

Of course the code at one point had properly defined foo. But then as the months go on things change, and we never retried that print(), so it's embarrassing when it finally triggers. Routine static analysis with ruff, pyright, or mypy on each git push can help to surface such a "whoops!".

lint

It is customary to use LOGGER.exception() when reporting on some exception e, in order to reveal the call site via the displayed stack trace.

No need for an f-string when there's no { } params, such as f"foo = {foo}"

Prefer e.g. "hello" over f"hello"

\$\endgroup\$
2
\$\begingroup\$

The previous answer was thorough, but here are some additional suggestions.

DRY

Your LOGGER info and debug message string are very similar to each other. You could use a variable to store the info string, then use the variable in the debug string:

info = f"{name_test}  1.0.0: {'Work' if works_100 else 'Not work'}"
LOGGER.info(info)
LOGGER.debug(f"{info} | Out: {out_100} | Err: {err_100}")

docstrings

It is great that you included docstrings with some of your functions, but you used different quotes for both: """ and '''. it would be better to be consistent and use the same quotes.

Comments

Comments are intended to describe the code, but comments like these seem to be notes to yourself about a future enhancement:

# Think need record errors some other way. In future will change.

You can keep a record of this in another text file in your version control system, then delete the comments.

\$\endgroup\$

You must log in to answer this question.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.