github · serenac5a27ed9eca9

# This script provides a platform-independent way of making the jupyter-book call (used in pyproject.toml)

folder = Path(__file__).parent

toc_file = folder / "_toc.yml"

print(cmd)

os.system(cmd)

.. automodule:: {module_qualname}

   :members:

   :undoc-members:

def index_template(package_name: str, doc_references: Optional[List[str]] = None, text_prefix=""):

path = cls._get_persistence_path(hook)

            try:

                with open(path, "rb") as f:

            except Exception:

                return cls()

"""

    with open(path, encoding=SERENA_FILE_ENCODING) as f:

        yaml = _create_yaml(preserve_comments=True)

    if commented_map is None:  # ruamel returns None for empty documents, but we want an empty CommentedMap

        commented_map = CommentedMap()

    normalise_yaml_comments(commented_map, comment_normalisation)

for pattern in self.libRepo.ignoredFileGlobPatterns:

            for path in glob.glob(pattern, recursive=True):

                print("Restoring via git: %s" % path)

        # restore directories that were backed up

        for d in self.libRepo.fullyIgnoredUnversionedDirectories:

def gitCommitWithMessageFromFile(commitMsgFilename):

    if not os.path.exists(commitMsgFilename):

        raise FileNotFoundError(f"{commitMsgFilename} not found in {os.path.abspath(os.getcwd())}")

    os.unlink(commitMsgFilename)

if deploy:

        user = os.getenv("HADES_USER")

        assert user, "HADES_USER environment variable must be set to deploy news.json to Hades"

click.echo(f"Bumped version to {new_version}")

        os.system("git add -u")

        os.system(f'git commit -m "Release v{new_version}"')

def find_repo_root() -> Path:

self.restore_ignored_files(libRepo.rootPath)

        # make commit in lib repo

        gitCommitWithMessageFromFile(COMMIT_MSG_FILENAME)

        newSyncCommitIdLibRepo = call("git rev-parse HEAD").strip()

os.system("uv lock")

        click.echo(f"Bumped version to {new_version}")

        os.system("git add -u")

        os.system(f"git tag v{new_version}")

if not cls._is_safe_tar_member(member, target_dir):

                            log.error(f"Unsafe tar member detected (path traversal): {member.name}")

                            return False

            elif archive_type == "zip":

                with zipfile.ZipFile(archive_path, "r") as zip_ref:

# Install nodejs

ENV NVM_VERSION=0.40.3

ENV NODE_VERSION=22.18.0

# standard location

ENV NVM_DIR=/root/.nvm

RUN . "$NVM_DIR/nvm.sh" && nvm install ${NODE_VERSION}

ENV PATH="${PATH}:/root/.local/bin"

# Install the latest version of uv

# Install Rust and rustup for rust-analyzer support (minimal profile)

ENV RUSTUP_HOME=/usr/local/rustup

ENV RUSTUP_HOME=/usr/local/rustup

ENV CARGO_HOME=/usr/local/cargo

ENV PATH="${CARGO_HOME}/bin:${PATH}"

    --default-toolchain stable \

    --profile minimal \

    && rustup component add rust-analyzer

FROM python:3.11-slim AS base

SHELL ["/bin/bash", "-c"]

# Set environment variables to make Python print directly to the terminal and avoid .pyc files.

ENV PYTHONUNBUFFERED=1

ENV PYTHONDONTWRITEBYTECODE=1

# Install system dependencies required for package manager and build tools.

# sudo, wget, zip needed for some assistants, like junie

RUN apt-get update && apt-get install -y --no-install-recommends \

    curl \

    build-essential \

    git \

    ssh \

execute('git add %s %s' % (self.SYNC_COMMIT_ID_FILE_LIB_REPO, self.SYNC_COMMIT_ID_FILE_THIS_REPO))

        execute(f'git commit -m "{self.SYNC_COMMIT_MESSAGE % self.libRepo.name} (pull)"')

              f"'{self.branch}'\nand get your branch merged into develop (issuing a pull request where appropriate)")

    def push(self, libRepo: "LibRepo"):

        """

result_message = self._rename_memory(request_rename_memory)

                return {"status": "success", "message": result_message}

            except Exception as e:

        @self._app.route("/get_serena_config", methods=["GET"])

        def get_serena_config() -> dict[str, Any]:

self._save_serena_config(request_save_config)

                return {"status": "success", "message": "Serena config saved successfully"}

            except Exception as e:

        @self._app.route("/queued_task_executions", methods=["GET"])

        def get_queued_executions() -> dict[str, Any]:

result = self._get_memory(request_get_memory)

                return result.model_dump()

            except Exception as e:

        @self._app.route("/save_memory", methods=["POST"])

        def save_memory() -> dict[str, str]:

response = [QueuedExecution.from_task_info(task_info).model_dump() for task_info in current_executions]

                return {"queued_executions": response, "status": "success"}

            except Exception as e:

        @self._app.route("/cancel_task_execution", methods=["POST"])

        def cancel_task_execution() -> dict[str, Any]:

result = self._get_serena_config()

                return result.model_dump()

            except Exception as e:

        @self._app.route("/save_serena_config", methods=["POST"])

        def save_serena_config() -> dict[str, str]:

response = QueuedExecution.from_task_info(last_execution_info).model_dump() if last_execution_info is not None else None

                return {"last_execution": response, "status": "success"}

            except Exception as e:

        @self._app.route("/fetch_unread_news", methods=["GET"])

        def fetch_unread_news() -> dict[str, dict[str, str] | str]:

self._delete_memory(request_delete_memory)

                return {"status": "success", "message": f"Memory {request_delete_memory.memory_name} deleted successfully"}

            except Exception as e:

        @self._app.route("/rename_memory", methods=["POST"])

        def rename_memory() -> dict[str, str]:

self._save_memory(request_save_memory)

                return {"status": "success", "message": f"Memory {request_save_memory.memory_name} saved successfully"}

            except Exception as e:

        @self._app.route("/delete_memory", methods=["POST"])

        def delete_memory() -> dict[str, str]:

self._remove_language(request_remove_language)

                return {"status": "success", "message": f"Language {request_remove_language.language} removed successfully"}

            except Exception as e:

        @self._app.route("/get_memory", methods=["POST"])

        def get_memory() -> dict[str, Any]:

self._read_news.mark_read(news_snippet_id)

                return {"status": "success", "message": f"Marked news snippet {news_snippet_id} as read"}

            except Exception as e:

    def _get_log_messages(self, request_log: RequestLog) -> ResponseLog:

        messages = self._memory_log_handler.get_log_messages(from_idx=request_log.start_idx)

return {"path": path, "results": search_results, "error": None}

        except Exception as e:

            log.debug(f"Error processing {path}: {e}")

    # Execute in parallel using joblib

    results = Parallel(

"message": f"Task with id {escape(request_data.get('task_id'))} not found, maybe execution was already finished",

                }

            except Exception as e:

        @self._app.route("/last_execution", methods=["GET"])

        def get_last_execution() -> dict[str, Any]:

self._add_language(request_add_language)

                return {"status": "success", "message": f"Language {request_add_language.language} added successfully"}

            except Exception as e:

        @self._app.route("/remove_language", methods=["POST"])

        def remove_language() -> dict[str, str]:

# Signal children first

            for child in parent.children(recursive=True):

                try:

                except (psutil.NoSuchProcess, psutil.AccessDenied, Exception):

                    pass

            # Then signal the parent

            try:

# Clean up temp directory

            if os.path.exists(temp_dir):

                try:

                except Exception:

                    pass

            return False

children = parent.children(recursive=True)

                for child in children:

                    try:

                    except (psutil.NoSuchProcess, psutil.AccessDenied):

                        pass

                psutil.wait_procs(children, timeout=2)

                for child in children:

                    try:

self._log_buffer.append(msg)

                for callback in self._emit_callbacks:

                    try:

                    except:

                        pass

                self._log_queue.task_done()

            except queue.Empty:

                continue

with open(gemfile_path) as f:

                    content = f.read().lower()

                    if "gem 'rails'" in content or 'gem "rails"' in content:

            except Exception:

                pass

        return False