From 1c45049bb83455c4ab8af8a13a244d53b6e90bf7 Mon Sep 17 00:00:00 2001 From: skyflow-vivek <121149418+skyflow-vivek@users.noreply.github.com> Date: Tue, 16 Sep 2025 22:08:12 +0530 Subject: [PATCH 1/2] Merge pull request #199 from skyflowapi/skyflow-vivek/SK-2296-use-sessions-for-insert SK-2296 Use sessions for insert method --- .github/workflows/internal-release.yml | 22 ++++++ .github/workflows/shared-build-and-deploy.yml | 74 +++++++++++++++++++ ci-scripts/bump_version.sh | 13 ++-- skyflow/_utils.py | 1 + skyflow/errors/_skyflow_errors.py | 1 + skyflow/vault/_client.py | 69 +++++++++++------ 6 files changed, 149 insertions(+), 31 deletions(-) create mode 100644 .github/workflows/internal-release.yml create mode 100644 .github/workflows/shared-build-and-deploy.yml diff --git a/.github/workflows/internal-release.yml b/.github/workflows/internal-release.yml new file mode 100644 index 00000000..cafdb9c9 --- /dev/null +++ b/.github/workflows/internal-release.yml @@ -0,0 +1,22 @@ +name: Internal Release + +on: + push: + tags-ignore: + - '*.*' + paths-ignore: + - "setup.py" + - "*.yml" + - "*.md" + - "skyflow/version.py" + - "samples/**" + branches: + - release/* + +jobs: + build-and-deploy: + uses: ./.github/workflows/shared-build-and-deploy.yml + with: + ref: ${{ github.ref_name }} + tag: 'internal' + secrets: inherit \ No newline at end of file diff --git a/.github/workflows/shared-build-and-deploy.yml b/.github/workflows/shared-build-and-deploy.yml new file mode 100644 index 00000000..cc6ac048 --- /dev/null +++ b/.github/workflows/shared-build-and-deploy.yml @@ -0,0 +1,74 @@ +name: Shared Build and Deploy + +on: + workflow_call: + inputs: + ref: + description: 'Git reference to use (e.g., main or branch name)' + required: true + type: string + + tag: + description: 'Release Tag' + required: true + type: string + +jobs: + build-and-deploy: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v2 + with: + fetch-depth: 0 + + - uses: actions/setup-python@v2 + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install setuptools wheel twine + - name: Resolve Branch for the Tagged Commit + id: resolve-branch + if: ${{ inputs.tag == 'beta' || inputs.tag == 'public' }} + run: | + TAG_COMMIT=$(git rev-list -n 1 ${{ github.ref_name }}) + BRANCH_NAME=$(git branch -r --contains $TAG_COMMIT | grep -o 'origin/.*' | sed 's|origin/||' | head -n 1) + if [ -z "$BRANCH_NAME" ]; then + echo "Error: Could not resolve branch for the tag." + exit 1 + fi + echo "Resolved Branch Name: $BRANCH_NAME" + echo "branch_name=$BRANCH_NAME" >> $GITHUB_ENV + - name: Get Previous tag + id: previoustag + uses: WyriHaximus/github-action-get-previous-tag@v1 + with: + fallback: 1.0.0 + + - name: Bump Version + run: | + chmod +x ./ci-scripts/bump_version.sh + if ${{ inputs.tag == 'internal' }}; then + ./ci-scripts/bump_version.sh "${{ steps.previoustag.outputs.tag }}" "$(git rev-parse --short "$GITHUB_SHA")" + else + ./ci-scripts/bump_version.sh "${{ steps.previoustag.outputs.tag }}" + fi + - name: Commit changes + run: | + git config user.name "${{ github.actor }}" + git config user.email "${{ github.actor }}@users.noreply.github.com" + git add setup.py + git add skyflow/version.py + if [[ "${{ inputs.tag }}" == "internal" ]]; then + VERSION="${{ steps.previoustag.outputs.tag }}.dev0+$(git rev-parse --short $GITHUB_SHA)" + COMMIT_MESSAGE="[AUTOMATED] Private Release $VERSION" + git commit -m "$COMMIT_MESSAGE" + git push origin ${{ github.ref_name }} -f + fi + - name: Build and Publish to JFrog Artifactory + if: ${{ inputs.tag == 'internal' }} + env: + TWINE_USERNAME: ${{ secrets.JFROG_USERNAME }} + TWINE_PASSWORD: ${{ secrets.JFROG_PASSWORD }} + run: | + python setup.py sdist bdist_wheel + twine upload --repository-url https://prekarilabs.jfrog.io/artifactory/api/pypi/skyflow-python/ dist/* \ No newline at end of file diff --git a/ci-scripts/bump_version.sh b/ci-scripts/bump_version.sh index b0a57a9e..c8f2b9e9 100755 --- a/ci-scripts/bump_version.sh +++ b/ci-scripts/bump_version.sh @@ -1,22 +1,19 @@ Version=$1 SEMVER=$Version - if [ -z $2 ] then echo "Bumping package version to $1" - sed -E "s/current_version = .+/current_version = \'$SEMVER\'/g" setup.py > tempfile && cat tempfile > setup.py && rm -f tempfile sed -E "s/SDK_VERSION = .+/SDK_VERSION = \'$SEMVER\'/g" skyflow/version.py > tempfile && cat tempfile > skyflow/version.py && rm -f tempfile - echo -------------------------- echo "Done, Package now at $1" else - echo "Bumping package version to $1-dev.$2" + echo "Bumping package version to $1.dev0+$2" - sed -E "s/current_version = .+/current_version = \'$SEMVER-dev.$2\'/g" setup.py > tempfile && cat tempfile > setup.py && rm -f tempfile - sed -E "s/SDK_VERSION = .+/SDK_VERSION = \'$SEMVER-dev.$2\'/g" skyflow/version.py > tempfile && cat tempfile > skyflow/version.py && rm -f tempfile + sed -E "s/current_version = .+/current_version = \'$SEMVER.dev0+$2\'/g" setup.py > tempfile && cat tempfile > setup.py && rm -f tempfile + sed -E "s/SDK_VERSION = .+/SDK_VERSION = \'$SEMVER.dev0+$2\'/g" skyflow/version.py > tempfile && cat tempfile > skyflow/version.py && rm -f tempfile echo -------------------------- - echo "Done, Package now at $1-dev.$2" -fi + echo "Done, Package now at $1.dev0+$2" +fi \ No newline at end of file diff --git a/skyflow/_utils.py b/skyflow/_utils.py index 83bf54a6..1e1b7109 100644 --- a/skyflow/_utils.py +++ b/skyflow/_utils.py @@ -53,6 +53,7 @@ def log_error(message: str, interface: str): class InfoMessages(Enum): INITIALIZE_CLIENT = "Initializing skyflow client" CLIENT_INITIALIZED = "Initialized skyflow client successfully" + CLOSING_SESSION = "Closing the session" VALIDATE_INSERT_RECORDS = "Validating insert records" VALIDATE_DETOKENIZE_INPUT = "Validating detokenize input" VALIDATE_GET_BY_ID_INPUT = "Validating getByID input" diff --git a/skyflow/errors/_skyflow_errors.py b/skyflow/errors/_skyflow_errors.py index 2e792812..ba52c4e5 100644 --- a/skyflow/errors/_skyflow_errors.py +++ b/skyflow/errors/_skyflow_errors.py @@ -16,6 +16,7 @@ class SkyflowErrorCodes(Enum): class SkyflowErrorMessages(Enum): API_ERROR = "Server returned status code %s" + NETWORK_ERROR = "Network error occurred: %s" FILE_NOT_FOUND = "File at %s not found" FILE_INVALID_JSON = "File at %s is not in JSON format" diff --git a/skyflow/vault/_client.py b/skyflow/vault/_client.py index e21dcbab..97979505 100644 --- a/skyflow/vault/_client.py +++ b/skyflow/vault/_client.py @@ -5,6 +5,7 @@ import types import requests import asyncio +from requests.adapters import HTTPAdapter from skyflow.vault._insert import getInsertRequestBody, processResponse, convertResponse from skyflow.vault._update import sendUpdateRequests, createUpdateResponseBody from skyflow.vault._config import Configuration, ConnectionConfig, DeleteOptions, DetokenizeOptions, GetOptions, InsertOptions, UpdateOptions, QueryOptions @@ -36,49 +37,71 @@ def __init__(self, config: Configuration): raise SkyflowError(SkyflowErrorCodes.INVALID_INPUT, SkyflowErrorMessages.TOKEN_PROVIDER_ERROR.value % ( str(type(config.tokenProvider))), interface=interface) + self._create_session() self.vaultID = config.vaultID self.vaultURL = config.vaultURL.rstrip('/') self.tokenProvider = config.tokenProvider self.storedToken = '' log_info(InfoMessages.CLIENT_INITIALIZED.value, interface=interface) + + def _create_session(self): + self.session = requests.Session() + adapter = HTTPAdapter(pool_connections=1, pool_maxsize=25, pool_block=True) + self.session.mount("https://", adapter) + + def __del__(self): + if (self.session is not None): + log_info(InfoMessages.CLOSING_SESSION.value, interface=InterfaceName.CLIENT.value) + self.session.close() + self.session = None + + def _get_session(self): + if (self.session is None): + self._create_session() + return self.session def insert(self, records: dict, options: InsertOptions = InsertOptions()): + max_retries = 1 interface = InterfaceName.INSERT.value log_info(InfoMessages.INSERT_TRIGGERED.value, interface=interface) self._checkConfig(interface) - jsonBody = getInsertRequestBody(records, options) requestURL = self._get_complete_vault_url() - self.storedToken = tokenProviderWrapper( - self.storedToken, self.tokenProvider, interface) - headers = { - "Authorization": "Bearer " + self.storedToken, - "sky-metadata": json.dumps(getMetrics()) - } - max_retries = 3 - # Use for-loop for retry logic, avoid code repetition - for attempt in range(max_retries+1): + + for attempt in range(max_retries + 1): try: - # If jsonBody is a dict, use json=, else use data= - response = requests.post(requestURL, data=jsonBody, headers=headers) + self.storedToken = tokenProviderWrapper( + self.storedToken, self.tokenProvider, interface) + headers = { + "Authorization": "Bearer " + self.storedToken, + "sky-metadata": json.dumps(getMetrics()), + } + response = self._get_session().post( + requestURL, + data=jsonBody, + headers=headers, + ) processedResponse = processResponse(response) result, partial = convertResponse(records, processedResponse, options) if partial: log_error(SkyflowErrorMessages.BATCH_INSERT_PARTIAL_SUCCESS.value, interface) - raise SkyflowError(SkyflowErrorCodes.PARTIAL_SUCCESS, SkyflowErrorMessages.BATCH_INSERT_PARTIAL_SUCCESS.value, result, interface=interface) - if 'records' not in result: + elif 'records' not in result: log_error(SkyflowErrorMessages.BATCH_INSERT_FAILURE.value, interface) - raise SkyflowError(SkyflowErrorCodes.SERVER_ERROR, SkyflowErrorMessages.BATCH_INSERT_FAILURE.value, result, interface=interface) - log_info(InfoMessages.INSERT_DATA_SUCCESS.value, interface) + else: + log_info(InfoMessages.INSERT_DATA_SUCCESS.value, interface) return result - except Exception as err: + except requests.exceptions.ConnectionError as err: if attempt < max_retries: - continue - else: - if isinstance(err, SkyflowError): - raise err - else: - raise SkyflowError(SkyflowErrorCodes.SERVER_ERROR, f"Error occurred: {err}", interface=interface) + continue + raise SkyflowError( + SkyflowErrorCodes.SERVER_ERROR, + SkyflowErrorMessages.NETWORK_ERROR.value % str(err), + interface=interface + ) + except SkyflowError as err: + if err.code != SkyflowErrorCodes.SERVER_ERROR or attempt >= max_retries: + raise err + continue def detokenize(self, records: dict, options: DetokenizeOptions = DetokenizeOptions()): interface = InterfaceName.DETOKENIZE.value From 9a9ab2f5f3ad15fd3e1bb7350d1b11ce9bae7b09 Mon Sep 17 00:00:00 2001 From: skyflow-shravan Date: Tue, 16 Sep 2025 16:38:32 +0000 Subject: [PATCH 2/2] [AUTOMATED] Private Release 1.15.4.dev0+1c45049 --- setup.py | 2 +- skyflow/version.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/setup.py b/setup.py index 323fa31d..c83828ab 100644 --- a/setup.py +++ b/setup.py @@ -6,7 +6,7 @@ if sys.version_info < (3, 7): raise RuntimeError("skyflow requires Python 3.7+") -current_version = '1.15.4' +current_version = '1.15.4.dev0+1c45049' setup( name='skyflow', diff --git a/skyflow/version.py b/skyflow/version.py index 5c7ae5de..fa38ca14 100644 --- a/skyflow/version.py +++ b/skyflow/version.py @@ -1 +1 @@ -SDK_VERSION = '1.15.4' \ No newline at end of file +SDK_VERSION = '1.15.4.dev0+1c45049' \ No newline at end of file