How to use the aiofiles.os function in aiofiles

To help you get started, we’ve selected a few aiofiles examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github huge-success / sanic / tests / test_response.py View on Github external
async def file_route(request, filename):
        file_path = os.path.join(static_file_directory, filename)
        file_path = os.path.abspath(unquote(file_path))
        stats = await async_os.stat(file_path)
        headers = dict()
        headers["Accept-Ranges"] = "bytes"
        headers["Content-Length"] = str(stats.st_size)
        if request.method == "HEAD":
            return HTTPResponse(
                headers=headers,
                content_type=guess_type(file_path)[0] or "text/plain",
            )
        else:
            return file(
                file_path,
                headers=headers,
                mime_type=guess_type(file_path)[0] or "text/plain",
            )
github huge-success / sanic / tests / test_response.py View on Github external
async def file_route(request, filename):
        file_path = os.path.join(static_file_directory, filename)
        file_path = os.path.abspath(unquote(file_path))
        headers = dict()
        headers["Accept-Ranges"] = "bytes"
        if request.method == "HEAD":
            # Return a normal HTTPResponse, not a
            # StreamingHTTPResponse for a HEAD request
            stats = await async_os.stat(file_path)
            headers["Content-Length"] = str(stats.st_size)
            return HTTPResponse(
                headers=headers,
                content_type=guess_type(file_path)[0] or "text/plain",
            )
        else:
            return file_stream(
                file_path,
                chunk_size=32,
                headers=headers,
                mime_type=guess_type(file_path)[0] or "text/plain",
            )
github Tinche / aiofiles / tests / test_os.py View on Github external
def test_stat():
    """Test the stat call."""
    filename = join(dirname(__file__), 'resources', 'test_file1.txt')

    stat_res = yield from aiofiles.os.stat(filename)

    assert stat_res.st_size == 10
github omarryhan / aiogoogle / aiogoogle / sessions / aiohttp_session.py View on Github external
async def _get_file_size(full_file_path):
    stat = await async_os.stat(full_file_path)
    return stat.st_size
github N-Coder / studip-fuse / studip_fuse / launcher / aioimpl / asyncio / aiohttp_client.py View on Github external
import aiohttp
import attr
from aiohttp import ClientRequest, hdrs, helpers
from async_exit_stack import AsyncExitStack
from async_generator import async_generator, asynccontextmanager, yield_
from async_lru import alru_cache
from oauthlib.oauth1 import Client as OAuth1Client
from pyrsistent import freeze
from yarl import URL

from studip_fuse.studipfs.api.aiobase import BaseHTTPClient
from studip_fuse.studipfs.api.aiointerface import Download

log = logging.getLogger(__name__)

async_stat = aiofiles.os.stat
async_utime = aiofiles.os.wrap(os.utime)


class AuthenticatedClientRequest(ClientRequest):
    def update_auth(self, auth):
        if auth is None:
            auth = self.auth
        if auth is None:
            return

        if isinstance(auth, helpers.BasicAuth):
            self.headers[hdrs.AUTHORIZATION] = auth.encode()
        elif isinstance(auth, OAuth1Client):
            url, headers, _ = auth.sign(
                str(self.url), str(self.method), None, self.headers
            )
github N-Coder / studip-fuse / studip_fuse / cache / cached_session.py View on Github external
async def has_cached_download(self, studip_file, local_dest=None):
        if not local_dest:
            local_dest = os.path.join(self.cache_dir, studip_file.id)

        # check integrity of existing paths (file with id exists, same size, same change date) and reuse them
        timestamp = time.mktime(studip_file.changed.timetuple())
        try:
            stat = await aio_os.stat(local_dest)
            return S_ISREG(stat.st_mode) and stat.st_size == studip_file.size and stat.st_mtime == timestamp
        except FileNotFoundError:
            return False