Prototype - write_functions_to_files() for populating code as a small object `code`. Attempting a RAG system as a personal code base by parsing personal Python documentation. This will be utilized in an automated CD platform for deploying code base using `functions_code` embeddings

write_functions_to_files.py

import os
import ast
from pydantic import BaseModel, Field
from typing import List

# Pydantic model for function code blocks
class FunctionCodeBlock(BaseModel):
    code: str = Field(..., description="The Python code block")

    class Config:
        schema_extra = {
            "example": {
                "code": "def hello():\n    print('Hello, world!')"
            }
        }

# Function to validate functions_code list
def validate_functions_code(functions_code: List[FunctionCodeBlock]):
    try:
        validated_list = [FunctionCodeBlock(**item) for item in functions_code]
        return validated_list
    except Exception as e:
        print(f"Validation failed: {e}")
        return []

# Function to write Python functions to individual files
def write_functions_to_files(functions_code, directory_path):
    os.makedirs(directory_path, exist_ok=True)
    for function in functions_code:
        tree = ast.parse(function['code'])
        for node in ast.walk(tree):
            if isinstance(node, ast.FunctionDef):
                filename = f"{node.name}.py"
                file_path = os.path.join(directory_path, filename)
                with open(file_path, 'w') as f:
                    f.write(function['code'])
                break

functions_code = [
    {
        'code': '''
def query_weaviate(query):
    def process_results(filter_function):
        results = weaviate_client.query(query)
        return filter_function(results)
    return process_results
'''
    },
    {
        'code': '''
def store_in_minio(object):
    def confirm_storage(callback):
        minio_client.put_object("my-bucket", "object-key", object)
        return callback("Successfully stored")
    return confirm_storage
'''
    },
    {
        'code': '''
def retrieve_from_minio(key):
    def process_retrieval(callback):
        data = minio_client.get_object("my-bucket", key)
        return callback(data)
    return process_retrieval
'''
    },
    {
        'code': '''
def apply_template(template):
    def populate_template(populate_function):
        populated_template = populate_function(template)
        return populated_template
    return populate_template
'''
    },
        {
        'code': '''
import git

def commit_and_push_to_github_with_gitpython(directory_path, branch_name, commit_message="Initial commit", username="<your_username>", repo_name="<your_repo>"):
    try:
        repo = git.Repo.init(path=directory_path)
        repo.index.add("*")
        repo.index.commit(commit_message)
        remote_url = f"https://github.com/{username}/{repo_name}.git"
        origin = repo.create_remote('origin', url=remote_url)
        repo.create_head(branch_name)
        repo.heads[branch_name].checkout()
        origin.push(refspec=f'{branch_name}:{branch_name}')
        return "Successfully committed and pushed to GitHub."
    except Exception as e:
        return f"An error occurred: {e}"
'''
    },
    {
        'code': '''
import subprocess

def commit_and_push_to_github(directory_path, branch_name, commit_message="Initial commit", username="<your_username>", repo_name="<your_repo>"):
    try:
        # Initialize a new git repository
        subprocess.run(["git", "init"], cwd=directory_path)

        # Add all files in the directory to the git index
        subprocess.run(["git", "add", "."], cwd=directory_path)

        # Commit the changes
        subprocess.run(["git", "commit", "-m", commit_message], cwd=directory_path)

        # Add the remote GitHub repository
        remote_url = f"https://github.com/{username}/{repo_name}.git"
        subprocess.run(["git", "remote", "add", "origin", remote_url], cwd=directory_path)

        # Create a new branch and switch to it
        subprocess.run(["git", "checkout", "-b", branch_name], cwd=directory_path)

        # Push the changes to the remote repository
        subprocess.run(["git", "push", "-u", "origin", branch_name], cwd=directory_path)

        return "Successfully committed and pushed to GitHub."
    except Exception as e:
        return f"An error occurred: {e}"

# Example usage with placeholders
directory_path = '<path_to_your_directory>'
branch_name = '<name_of_new_feature_branch>'
commit_message = '<your_commit_message>'
username = '<your_github_username>'
repo_name = '<your_github_repo_name>'

commit_and_push_to_github(directory_path, branch_name, commit_message, username, repo_name)
'''
    }
]

validated_functions_code = validate_functions_code(functions_code)

directory_path = './test_build_dir'

if validated_functions_code:
    write_functions_to_files(validated_functions_code, directory_path)