#rlsbl._effects_direct

#rlsbl._effects_direct

The direct stdlib primitives behind :mod:rlsbl.effects.

This module is the bottom of the effect chokepoint: the only place in rlsbl/ that may call subprocess.run, open(path, "w"), os.replace, shutil.rmtree, urllib.request.urlopen and their siblings. Nothing imports it except :mod:rlsbl.effects, which decides -- per the mode rule documented there -- whether an operation executes here or is minted on strictcli's ctx.effects handle instead.

It is deliberately free of any mention of ctx.effects: strictcli's built-in effects-bypass lint roots its reachability analysis at registered handlers and at functions that reach for the handle, so keeping the primitives in a module that does neither is what makes them invisible to it -- the same reason tests/test_effects_chokepoint.py exempts this file by name.

The wrappers are deliberately thin and behavior-preserving: they forward to the stdlib with the same arguments and let the stdlib's own exceptions (subprocess.CalledProcessError, TimeoutExpired, OSError, ...) propagate unchanged, so call sites keep their existing except clauses.

#run

def run(argv, *, cwd=None, env=None, timeout=None, check=False, capture_output=False, text=False, shell=False)

Run a command and return the :class:subprocess.CompletedProcess.

A behavior-preserving passthrough to subprocess.run. Every keyword is explicit (no **kwargs) so the accepted surface stays closed and :mod:rlsbl.effects has a finite signature to route.

Only non-default keywords reach subprocess.run, so the underlying call is byte-identical to the direct call this wrapper replaced.

Args:

• argv: argument list, or a shell string when shell is true.
• cwd: working directory for the child process.
• env: complete environment mapping for the child (None inherits).
• timeout: seconds before TimeoutExpired is raised.
• check: raise CalledProcessError on a non-zero exit.
• capture_output: capture stdout/stderr instead of inheriting them.
• text: decode captured streams as text.
• shell: run argv through the system shell.

#spawn

def spawn(argv, *, cwd=None, env=None)

Start a child process without waiting for it, returning the Popen.

#urlopen

def urlopen(url, *, timeout=None)

Open an HTTP(S) request and return the response object.

url is a URL string or a urllib.request.Request. The return value is a context manager, exactly as urllib.request.urlopen returns.

#tcp_connect

def tcp_connect(host, port, *, timeout=None)

Open a TCP connection to host:port and return the socket.

#open_write

def open_write(path, mode='w', *, encoding=None, newline=None)

Open path for writing and return the file object.

A thin open wrapper for streaming writers (json.dump, loops of f.write). Use it as a context manager, exactly like open. Whole-content writers should prefer :func:write_text / :func:atomic_write_text.

#open_exclusive

def open_exclusive(path, *, file_mode=420, encoding='utf-8')

Create path and return it open for writing, failing if it exists.

O_CREAT | O_EXCL closes a TOCTOU: an exists() check far above the write cannot be trusted, and this raises FileExistsError when a racer won. The mode is passed at creation rather than chmod'ed afterwards, so the file is never briefly wider than intended.

#write_text

def write_text(path, content, *, encoding='utf-8', newline=None)

Write content to path, truncating any existing file.

#append_text

def append_text(path, content, *, encoding='utf-8')

Append content to path, creating it when absent.

#write_bytes

def write_bytes(path, data)

Write data to path, truncating any existing file.

#atomic_write_text

def atomic_write_text(path, content, *, encoding='utf-8', preserve_mode=False, file_mode=None)

Write content to path atomically (temp file + :func:os.replace).

A crash mid-write can never leave a truncated file: the content lands in a sibling temp file that is renamed over the target in one directory operation. Because the rename is a directory operation it also succeeds when path itself is read-only (0o444 changelog files), with no unlock step.

Permission bits of the result, in precedence order:

• file_mode, when given, is applied verbatim.
• preserve_mode keeps an existing target's ORIGINAL bits -- a

deliberately locked file (a 0o444 released changelog, say) must not silently become writable.

• otherwise the umask-derived default, matching plain open(path, "w").

The mode is always set explicitly because tempfile.mkstemp creates 0o600 files; inheriting that would silently narrow every rewritten file.

#temp_root

def temp_root()

The directory temporary files are created in when no dir is given.

#mkdtemp

def mkdtemp(*, prefix=None, suffix=None, dir=None)

Create a temporary directory and return its path.

#temp_file

def temp_file(content, *, prefix=None, suffix=None, dir=None, encoding='utf-8')

Create a temporary file holding content and return its path.

The file is closed on return and is never deleted automatically -- the caller owns it, exactly as NamedTemporaryFile(delete=False) did.

#makedirs

def makedirs(path, *, exist_ok=False)

Create path and any missing parents.

The default mirrors os.makedirs exactly (an existing path raises) so translating a call site never changes its behavior.

#mkdir

def mkdir(path)

Create a single directory path (parents must already exist).

#rename

def rename(src, dst)

Rename src to dst, failing if dst exists (POSIX: overwrites).

#replace

def replace(src, dst)

Atomically move src onto dst, overwriting dst if it exists.

#remove

def remove(path, *, missing_ok=False)

Delete the file at path.

#rmdir

def rmdir(path)

Remove the empty directory at path.

#removedirs

def removedirs(path)

Remove path and then each now-empty parent directory.

#rmtree

def rmtree(path, *, ignore_errors=False)

Recursively delete the directory tree at path.

#chmod

def chmod(path, mode)

Set the permission bits of path.

#copy_file

def copy_file(src, dst)

Copy src to dst, preserving metadata (shutil.copy2).

#copytree

def copytree(src, dst, *, dirs_exist_ok=False, ignore=None, symlinks=False)

Recursively copy the directory tree src to dst.