| 1208 | f.close() |
| 1209 | |
| 1210 | async def _copy_in(self, copy_stmt, source, timeout): |
| 1211 | try: |
| 1212 | path = os.fspath(source) |
| 1213 | except TypeError: |
| 1214 | # source is not a path-like object |
| 1215 | path = None |
| 1216 | |
| 1217 | f = None |
| 1218 | reader = None |
| 1219 | data = None |
| 1220 | opened_by_us = False |
| 1221 | run_in_executor = self._loop.run_in_executor |
| 1222 | |
| 1223 | if path is not None: |
| 1224 | # a path |
| 1225 | f = await run_in_executor(None, open, path, 'rb') |
| 1226 | opened_by_us = True |
| 1227 | elif hasattr(source, 'read'): |
| 1228 | # file-like |
| 1229 | f = source |
| 1230 | elif isinstance(source, collections.abc.AsyncIterable): |
| 1231 | # assuming calling output returns an awaitable. |
| 1232 | # copy_in() is designed to handle very large amounts of data, and |
| 1233 | # the source async iterable is allowed to return an arbitrary |
| 1234 | # amount of data on every iteration. |
| 1235 | reader = source |
| 1236 | else: |
| 1237 | # assuming source is an instance supporting the buffer protocol. |
| 1238 | data = source |
| 1239 | |
| 1240 | if f is not None: |
| 1241 | # Copying from a file-like object. |
| 1242 | class _Reader: |
| 1243 | def __aiter__(self): |
| 1244 | return self |
| 1245 | |
| 1246 | async def __anext__(self): |
| 1247 | data = await run_in_executor(None, f.read, 524288) |
| 1248 | if len(data) == 0: |
| 1249 | raise StopAsyncIteration |
| 1250 | else: |
| 1251 | return data |
| 1252 | |
| 1253 | reader = _Reader() |
| 1254 | |
| 1255 | try: |
| 1256 | return await self._protocol.copy_in( |
| 1257 | copy_stmt, reader, data, None, None, timeout) |
| 1258 | finally: |
| 1259 | if opened_by_us: |
| 1260 | await run_in_executor(None, f.close) |
| 1261 | |
| 1262 | async def set_type_codec(self, typename, *, |
| 1263 | schema='public', encoder, decoder, |