ls -ld /tmp
# should be: drwxrwxrwt (1777)

chmod 1777 /tmp

mount | grep ' /tmp '

"mounts": ["type=tmpfs,target=/tmp"]

"runArgs": ["--tmpfs", "/tmp:exec,mode=1777"]

mkdir -p /workspaces/your-repo/.tmp && chmod 777 /workspaces/your-repo/.tmp
export TMPDIR=/workspaces/your-repo/.tmp

"containerEnv": {
  "TMPDIR": "/workspaces/your-repo/.tmp"
},
"postCreateCommand": "mkdir -p /workspaces/your-repo/.tmp && chmod 777 /workspaces/your-repo/.tmp"

from prefect import flow

flow = flow.from_source(
    source="/workspaces/your-repo",
    entrypoint="path/to/flow.py:my_flow",
)
flow.serve(name="my-deployment")

python -c "import tempfile; print(tempfile.gettempdir())"

mkdir -p /tmp/runner_storage && touch /tmp/runner_storage/test && echo OK

ls -ld /tmp

prefect version

curl -sf <http://host.docker.internal:4200/api/health> || echo "unreachable"

"runArgs": ["--add-host=host.docker.internal:host-gateway"]

import os
  os.environ["TMPDIR"] = "/workspaces/your-repo/.tmp"
  os.makedirs(os.environ["TMPDIR"], exist_ok=True)
  os.chmod(os.environ["TMPDIR"], 0o777)

curl -sf "$PREFECT_API_URL/health" || echo "unreachable"

from prefect import flow

  flow = flow.from_source(
      source="<https://github.com/PrefectHQ/prefect.git>",
      entrypoint="examples/flows/hello_world.py:hello_world_flow",
  )
  flow.serve(name="smoke-test")

rm -rf /tmp/runner_storage/*
  # or fix ownership if needed
  # sudo chown -R "$(id -u):$(id -g)" /tmp/runner_storage

export TMPDIR=/workspaces/your-repo/.tmp
  mkdir -p "$TMPDIR" && chmod 777 "$TMPDIR"

gitlab_repository_block.directories = ["workflows"]

import os
import subprocess
from pathlib import Path
from prefect import flow

# Set a safe temp base first
os.environ["TMPDIR"] = "/workspaces/uvd-workflow-orchestrator-example-prefect/.tmp"
Path(os.environ["TMPDIR"]).mkdir(parents=True, exist_ok=True)

# 1) Pre-pull into a path you control
repo_dir = Path("/workspaces/uvd-workflow-orchestrator-example-prefect/_src")
if repo_dir.exists():
    # start fresh to avoid stale perms
    subprocess.run(["rm", "-rf", str(repo_dir)], check=True)

# Use your preferred git auth here (token/ssh). Example with HTTPS:
subprocess.run([
    "git", "clone", "--depth", "1",
    "<https://gitlab.example.com/group/project.git>",
    str(repo_dir)
], check=True)

# 2) Serve from the local path
my_flow = flow.from_source(
    source=str(repo_dir),
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
my_flow.serve(name="private-gitlab-deploy")

import os
from pathlib import Path

os.environ["TMPDIR"] = "/workspaces/uvd-workflow-orchestrator-example-prefect/.tmp"
Path(os.environ["TMPDIR"]).mkdir(parents=True, exist_ok=True)

# Clean any stale runner storage
rs_base = Path(os.environ["TMPDIR"]) / "runner_storage"
if rs_base.exists():
    # remove stale dirs that may be owned by another user
    import shutil
    shutil.rmtree(rs_base, ignore_errors=True)

ls -ld /tmp/runner_storage \
       /tmp/runner_storage/38726217-2925-408a-95f6-e65d81524334 \
       /tmp/runner_storage/38726217-2925-408a-95f6-e65d81524334/gitlab-repository-prefectprototypegit/.git/objects/pack

import os
import asyncio
from pathlib import Path

# Make sure this is set before importing Prefect to control temp base
os.environ["TMPDIR"] = "/workspaces/your-repo/.tmp"
Path(os.environ["TMPDIR"]).mkdir(parents=True, exist_ok=True)

from prefect.runner.storage import BlockStorageAdapter
from prefect_gitlab import GitLabRepository, GitLabCredentials

async def main():
    repo = GitLabRepository.load("prefectprototypegit")
    creds = GitLabCredentials.load("mycredential")
    repo.credentials = creds

    storage = BlockStorageAdapter(block=repo, pull_interval=60)

    # Preflight: perform the checkout
    await storage.pull_code()

    # Inspect where it landed
    print("Pulled to:", storage.destination)
    # Optional: list files to verify
    import subprocess
    subprocess.run(["ls", "-la", str(storage.destination)], check=False)

asyncio.run(main())

from prefect import flow as flow_cls
flow = flow_cls.from_source(
    source=storage,
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
flow.serve(name="private-gitlab-deploy")

import os, asyncio, subprocess
from pathlib import Path
os.environ["TMPDIR"] = "/workspaces/your-repo/.tmp"
Path(os.environ["TMPDIR"]).mkdir(parents=True, exist_ok=True)

from prefect.runner.storage import create_storage_from_source

async def main():
    storage = create_storage_from_source("<https://github.com/PrefectHQ/prefect.git>")
    await storage.pull_code()
    print("Pulled to:", storage.destination)
    subprocess.run(["ls", "-la", str(storage.destination)], check=False)

asyncio.run(main())

rm -rf ${TMPDIR:-/tmp}/runner_storage/*

serve

pull_code()

.git/objects/pack/*

from prefect.runner.storage import BlockStorageAdapter
from prefect_gitlab import GitLabRepository, GitLabCredentials
import asyncio

async def main():
    repo = GitLabRepository.load("prefectprototypegit")
    repo.credentials = GitLabCredentials.load("mycredential")

    storage = BlockStorageAdapter(block=repo)
    await storage.pull_code()
    print("Pulled to:", storage.destination)

asyncio.run(main())

.git/objects/pack

${TMPDIR:-/tmp}/runner_storage/<uuid>/...

TMPDIR

export TMPDIR=/workspaces/your-repo/.tmp
    mkdir -p "$TMPDIR" && chmod 777 "$TMPDIR"

import shutil, pathlib
    if pathlib.Path(storage.destination).exists():
        shutil.rmtree(storage.destination, ignore_errors=True)
    await storage.pull_code()

.git

repo.directories = ["workflows"]  # adjust to your code dir(s)

chmod -R u+rwX,g+rwX /uvd-workflow-orchestrator-example-prefect/gitlab-repository-prefectprototypegit
    find /uvd-workflow-orchestrator-example-prefect/gitlab-repository-prefectprototypegit/.git/objects/pack -type f -exec chmod u+w,g+w {} +

from prefect import flow as flow_cls

  local_path = "/workspaces/your-repo/_src"  # your own clone target
  # ...run `git clone` here, or ensure it's clean...

  f = flow_cls.from_source(
      source=local_path,
      entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
  )
  f.serve(name="private-gitlab-deploy")

.git/objects/pack/*

print(storage.destination)

ls -ld

.git/objects/pack

flow.from_source(...).serve(...)

.git/objects/pack/*

.git/objects/pack/*

/tmp

rm -rf ${TMPDIR:-/tmp}/runner_storage/*
# If some dirs are root-owned, run with sudo or fix ownership:
# sudo chown -R "$(id -u):$(id -g)" ${TMPDIR:-/tmp}/runner_storage

import os, shutil
from pathlib import Path

os.environ["TMPDIR"] = "/workspaces/your-repo/.tmp"
Path(os.environ["TMPDIR"]).mkdir(parents=True, exist_ok=True)

# Optional: proactively clear prior runner storage for a clean start
shutil.rmtree(Path(os.environ["TMPDIR"]) / "runner_storage", ignore_errors=True)

.git

gitlab_repository_block.directories = ["workflows"]  # adjust to your code dirs

from prefect import flow as flow_cls

local_path = "/workspaces/your-repo/_src"
# ...git clone or clean this path first...

f = flow_cls.from_source(
    source=local_path,
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
f.serve(name="private-gitlab-deploy")

from prefect.runner.storage import BlockStorageAdapter
from prefect_gitlab import GitLabRepository, GitLabCredentials
import asyncio

async def main():
    repo = GitLabRepository.load("prefectprototypegit")
    repo.credentials = GitLabCredentials.load("mycredential")
    storage = BlockStorageAdapter(block=repo)
    await storage.pull_code()
    print("Pulled to:", storage.destination)

asyncio.run(main())

ls -l /uvd-workflow-orchestrator-example-prefect/gitlab-repository-prefectprototypegit/.git/objects/pack

repo = GitLabRepository.load("prefectprototypegit")
repo.credentials = GitLabCredentials.load("mycredential")
repo.directories = ["workflows"]  # adjust to your code dir(s)

f = flow.from_source(
    source=repo,
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
f.serve(name="private-gitlab-deploy")

from prefect.runner.storage import BlockStorageAdapter

repo = GitLabRepository.load("prefectprototypegit")
repo.credentials = GitLabCredentials.load("mycredential")

storage = BlockStorageAdapter(block=repo, pull_interval=None)  # no periodic re-copy

f = flow.from_source(
    source=storage,
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
f.serve(name="private-gitlab-deploy")

import os, shutil, tempfile
from pathlib import Path

os.environ["TMPDIR"] = "/uvd-workflow-orchestrator-example-prefect/.tmp"
Path(os.environ["TMPDIR"]).mkdir(parents=True, exist_ok=True)
shutil.rmtree(Path(tempfile.gettempdir()) / "runner_storage", ignore_errors=True)

# then construct your repo/block and serve as usual

f = flow.from_source(
    source="/uvd-workflow-orchestrator-example-prefect/_src",  # your own clone target
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
f.serve(name="private-gitlab-deploy")

import os
from prefect import flow as flow_cls
from prefect.runner.storage import GitRepository

# Example using a basic HTTPS URL with a token in env (avoid committing tokens!)
# For GitLab, username can be anything; many use 'oauth2' or your actual username.
GITLAB_TOKEN = os.environ["GITLAB_TOKEN"]
url = f"<https://oauth2:{GITLAB_TOKEN}@gitlab.com/your-group/your-repo.git>"

repo = GitRepository(
    url=url,
    directories=["workflows"],   # only copy your flow code; .git is not copied
    pull_interval=60,            # or None to disable periodic pulls
)

f = flow_cls.from_source(
    source=repo,
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
f.serve(name="private-gitlab-deploy")

from prefect.runner.storage import BlockStorageAdapter
from prefect_gitlab import GitLabRepository, GitLabCredentials
from prefect import flow as flow_cls

repo = GitLabRepository.load("prefectprototypegit")
repo.credentials = GitLabCredentials.load("mycredential")

storage = BlockStorageAdapter(block=repo, pull_interval=None)  # no periodic re-copy

f = flow_cls.from_source(
    source=storage,
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
f.serve(name="private-gitlab-deploy")

rm -rf "/uvd-workflow-orchestrator-example-prefect/.tmp/runner_storage/b2f7289f-38dc-47a8-9fb4-ac790676347d/uvd-workflow-orchestrator-example-prefect.git"

rm -rf "/uvd-workflow-orchestrator-example-prefect/.tmp/runner_storage"

import os, shutil, tempfile
from pathlib import Path

os.environ["TMPDIR"] = "/uvd-workflow-orchestrator-example-prefect/.tmp"
Path(os.environ["TMPDIR"]).mkdir(parents=True, exist_ok=True)

# Clear any stale repos to avoid remote mismatch / EACCES on .git/objects/pack
shutil.rmtree(Path(tempfile.gettempdir()) / "runner_storage", ignore_errors=True)

from prefect.runner.storage import BlockStorageAdapter
from prefect_gitlab import GitLabRepository, GitLabCredentials
from prefect import flow as flow_cls

repo = GitLabRepository.load("prefectprototypegit")
repo.credentials = GitLabCredentials.load("mycredential")

storage = BlockStorageAdapter(block=repo, pull_interval=None)  # no periodic re-copy

f = flow_cls.from_source(
    source=storage,
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
f.serve(name="private-gitlab-deploy")

git -C "/uvd-workflow-orchestrator-example-prefect/.tmp/runner_storage/b2f7289f-38dc-47a8-9fb4-ac790676347d/uvd-workflow-orchestrator-example-prefect.git" remote -v

import os
from prefect import flow as flow_cls
from prefect.runner.storage import GitRepository

os.environ["TMPDIR"] = "/uvd-workflow-orchestrator-example-prefect/.tmp"

# Use HTTPS with a token in env for GitLab
url = f"<https://oauth2:{os.environ['GITLAB_TOKEN']}@gitlab.com/your-group/your-repo.git>"

repo = GitRepository(
    url=url,
    directories=["workflows"],   # only copy your flow code; .git is not copied
    pull_interval=60,            # or None if you don't want periodic refresh
)

f = flow_cls.from_source(
    source=repo,
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
f.serve(name="private-gitlab-deploy")

repo.credentials = GitLabCredentials.load("mycredential")

credentials=None

from prefect_gitlab import GitLabRepository, GitLabCredentials

# Ensure your credentials block exists and has the token
creds = GitLabCredentials.load("mycredential")  # or create and save it if needed

# Attach creds to the repo block and SAVE it so the API has them
repo = GitLabRepository.load("prefectprototypegit")
repo.credentials = creds
repo.save("prefectprototypegit", overwrite=True)

# Sanity check: new load should show credentials present
assert GitLabRepository.load("prefectprototypegit").credentials is not None

import os
from pathlib import Path
from prefect import flow as flow_cls
from prefect.runner.storage import BlockStorageAdapter
from prefect_gitlab import GitLabRepository

# Optional but recommended: control temp base and start clean
os.environ["TMPDIR"] = "/uvd-workflow-orchestrator-example-prefect/.tmp"
Path(os.environ["TMPDIR"]).mkdir(parents=True, exist_ok=True)

repo = GitLabRepository.load("prefectprototypegit")

# Disable periodic recopy to avoid .git pack overwrites
storage = BlockStorageAdapter(block=repo, pull_interval=None)

f = flow_cls.from_source(
    source=storage,
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
f.serve(name="private-gitlab-deploy")

from prefect.runner.storage import GitRepository
from prefect import flow as flow_cls
import os

os.environ["TMPDIR"] = "/uvd-workflow-orchestrator-example-prefect/.tmp"

url = f"<https://oauth2:{os.environ['GITLAB_TOKEN']}@gitlab.com/your-group/your-repo.git>"

repo = GitRepository(
    url=url,
    directories=["workflows"],   # only copy your flow code; .git not copied
    pull_interval=60,            # or None to disable periodic refresh
)

f = flow_cls.from_source(
    source=repo,
    entrypoint="workflows/workflow_with_non_and_decorated_task.py:my_flow",
)
f.serve(name="private-gitlab-deploy")

rm -rf "/uvd-workflow-orchestrator-example-prefect/.tmp/runner_storage"

from prefect_gitlab import GitLabRepository
r = GitLabRepository.load("prefectprototypegit")
print("Has creds?", r.credentials is not None)