Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@add_repr
def read(path: str) -> Effect[HasFiles, OSError, str]:
"""
get an `Effect` that reads the content of a file as a str
Example:
>>> class Env:
... files = Files()
>>> read('foo.txt').run(Env())
'contents of foo.txt'
Args:
path: path to file
Return:
`Effect` that reads file located at `path`
"""
return get_environment(HasFiles).and_then(lambda env: env.files.read(path))
@add_repr
def write(path: str, content: str) -> Effect[HasFiles, OSError, None]:
"""
Get an `Effect` that writes to a file
Example:
>>> class Env:
... files = Files()
>>> write('foo.txt')('contents')\\
... .discard_and_then(read('foo.txt'))\\
... .run(Env())
'content of foo.txt'
Args:
path: the path of the file to be written
content the content to write
Return:
@add_repr
def exception(
msg: str, stack_info: bool = True, exc_info: Union[bool, ExcInfo] = True
) -> Depends[HasLogging, None]:
"""
Create an effect that calls built-in `logging.exception`
Example:
>>> class Env:
... logging = Logging()
>>> exception('hello!').run(Env())
ERROR:root:hello!
Args:
msg: the log message
stack_info: whether to include stack information in the log message
exc_info: whether to include exception info in the log message
@add_repr
def execute(query: str, *args: Any, timeout: float = None
) -> Effect[HasSQL, asyncpg.PostgresError, str]:
"""
Get an `Effect` that executes `query`
Example:
>>> class Env:
... sql = SQL('postgres://user@host/database')
>>> execute(
... 'INSERT INTO users(name, age) VALUES($1, $2)',
... 'bob',
... 32
... )(Env())
'INSERT 1'
Args:
@add_repr
def head(
url: str,
params: Mapping[str, Any] = None,
data: Union[aiohttp.FormData, bytes, dict] = None,
json: JSon = None,
cookies: Mapping = None,
headers: Mapping = None,
skip_auto_headers: Iterable[str] = None,
auth: aiohttp.BasicAuth = None,
allow_redirects: bool = True,
max_redirects: int = 10,
compress: bool = None,
chunked: int = None,
expect100: bool = False,
raise_for_status: bool = None,
read_until_eof: bool = True,
@add_repr
def critical(
msg: str, stack_info: bool = False, exc_info: Union[bool, ExcInfo] = False
) -> Depends[HasLogging, None]:
"""
Create an effect that calls built-in `logging.critical`
Example:
>>> class Env:
... logging = Logging()
>>> critical('hello!').run(Env())
CRITICAL:root:hello!
Args:
msg: the log message
stack_info: whether to include stack information in the log message
exc_info: whether to include exception info in the log message
@add_repr
def append_bytes(path: str, content: bytes) -> Effect[HasFiles, OSError, None]:
"""
Get an `Effect` that appends to a file
Example:
>>> class Env:
... files = Files()
>>> append_bytes('foo.txt')(b'content of foo.txt')\\
... .discard_and_then(read('foo.txt'))\\
... .run(Env())
'content of foo.txt'
Args:
path: the path of the file to be written
content the content to append
Return:
@add_repr
def patch(
url: str,
params: Mapping[str, Any] = None,
data: Union[aiohttp.FormData, bytes, dict] = None,
json: JSon = None,
cookies: Mapping = None,
headers: Mapping = None,
skip_auto_headers: Iterable[str] = None,
auth: aiohttp.BasicAuth = None,
allow_redirects: bool = True,
max_redirects: int = 10,
compress: bool = None,
chunked: int = None,
expect100: bool = False,
raise_for_status: bool = None,
read_until_eof: bool = True,
@add_repr
def fetch_one(query: str, *args: Any, timeout: float = None
) -> Effect[HasSQL, SQLError, Dict[str, Any]]:
"""
Get an `Effect` that executes `query` and returns the first \
result as a `Dict` or fails with `EmptyResultSetError` if the \
result set is empty
Example:
>>> class Env:
... sql = SQL('postgres://user@host/database')
>>> fetch_one('select * from users')(None)
Dict({'name': 'bob', 'age': 32})
Args:
query: query to execute
args: arguments for query
@add_repr
def delete(
url: str,
params: Mapping[str, Any] = None,
data: Union[aiohttp.FormData, bytes, dict] = None,
json: JSon = None,
cookies: Mapping = None,
headers: Mapping = None,
skip_auto_headers: Iterable[str] = None,
auth: aiohttp.BasicAuth = None,
allow_redirects: bool = True,
max_redirects: int = 10,
compress: bool = None,
chunked: int = None,
expect100: bool = False,
raise_for_status: bool = None,
read_until_eof: bool = True,