Spaces:
Sleeping
Sleeping
File size: 3,276 Bytes
acd4009 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 |
from dataclasses import dataclass, field
import requests
from exec_outcome import ExecOutcome
from typing import List, Optional, Union, Tuple
@dataclass
class ExtendedUnittest:
input: str
output: List[str] = field(default_factory=list)
result: Optional[str] = None
exec_outcome: Optional[ExecOutcome] = None
def json(self):
_json = self.__dict__
if self.exec_outcome is not None:
_json["exec_outcome"] = self.exec_outcome.name
return _json
@classmethod
def from_json(cls, _json):
return cls(
input=_json.get("input", ""),
output=_json.get("output", list()),
result=_json.get("result", None),
exec_outcome=_json.get("exec_outcome", None),
)
class EmptyValueError(Exception):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
class EmptyUnittestError(EmptyValueError):
pass
class EmptyLanguageError(EmptyValueError):
pass
class EmptySourceCodeError(EmptyValueError):
pass
class APICommunication:
_session: requests.Session
def __init__(self, server_url: str = "http://localhost:5000"):
self._session = requests.Session()
self.execute_code_url = f"{server_url}/api/execute_code"
self.get_runtimes_url = f"{server_url}/api/all_runtimes"
def __enter__(self):
return self
def __exit__(self, *args):
self._session.close()
def get_runtimes(self):
return self._session.get(self.get_runtimes_url).json()
def execute_code(
self,
language: str,
source_code: str,
unittests: List[dict],
limits: Optional[dict] = None,
block_network: bool = True,
stop_on_first_fail: bool = True,
use_sanitizer: bool = False,
compiler_program_name: Optional[str] = None,
compiler_flags: Optional[str] = None,
interpreter_cmd: Optional[str] = None,
interpreter_flags: Optional[str] = None,
sample_id: Optional[int] = None,
task_id: Union[str, int, None] = None,
) -> Tuple[List[ExtendedUnittest], Optional[int], Union[str, int, None]]:
if language is None:
raise EmptyLanguageError
if source_code is None:
raise EmptySourceCodeError
if unittests is None or len(unittests) == 0:
raise EmptyUnittestError
request_body = dict(
language=language,
source_code=source_code,
unittests=unittests,
limits=limits,
compile_cmd=compiler_program_name,
compile_flags=compiler_flags,
execute_cmd=interpreter_cmd,
execute_flags=interpreter_flags,
block_network=block_network,
stop_on_first_fail=stop_on_first_fail,
use_sanitizer=use_sanitizer,
)
json_response = self._session.post(
self.execute_code_url,
json=request_body,
headers={"Content-Type": "application/json"},
).json()
if "data" not in json_response:
return json_response, sample_id, task_id
return (
json_response["data"],
sample_id,
task_id,
)
|