我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用win32file._get_osfhandle()。
def test_basics(self): s = socket.socket() e = win32event.CreateEvent(None, 1, 0, None) win32file.WSAEventSelect(s, e, 0) self.assertEquals(win32file.WSAEnumNetworkEvents(s), {}) self.assertEquals(win32file.WSAEnumNetworkEvents(s, e), {}) self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, s, e, 3) self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, s, "spam") self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, "spam", e) self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, "spam") f = open("NUL") h = win32file._get_osfhandle(f.fileno()) self.assertRaises(win32file.error, win32file.WSAEnumNetworkEvents, h) self.assertRaises(win32file.error, win32file.WSAEnumNetworkEvents, s, h) try: win32file.WSAEnumNetworkEvents(h) except win32file.error, e: self.assertEquals(e.winerror, win32file.WSAENOTSOCK) try: win32file.WSAEnumNetworkEvents(s, h) except win32file.error, e: # According to the docs it would seem reasonable that # this would fail with WSAEINVAL, but it doesn't. self.assertEquals(e.winerror, win32file.WSAENOTSOCK)
def test_basics(self): s = socket.socket() e = win32event.CreateEvent(None, 1, 0, None) win32file.WSAEventSelect(s, e, 0) self.assertEquals(win32file.WSAEnumNetworkEvents(s), {}) self.assertEquals(win32file.WSAEnumNetworkEvents(s, e), {}) self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, s, e, 3) self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, s, "spam") self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, "spam", e) self.assertRaises(TypeError, win32file.WSAEnumNetworkEvents, "spam") f = open("NUL") h = win32file._get_osfhandle(f.fileno()) self.assertRaises(win32file.error, win32file.WSAEnumNetworkEvents, h) self.assertRaises(win32file.error, win32file.WSAEnumNetworkEvents, s, h) try: win32file.WSAEnumNetworkEvents(h) except win32file.error as e: self.assertEquals(e.winerror, win32file.WSAENOTSOCK) try: win32file.WSAEnumNetworkEvents(s, h) except win32file.error as e: # According to the docs it would seem reasonable that # this would fail with WSAEINVAL, but it doesn't. self.assertEquals(e.winerror, win32file.WSAENOTSOCK)
def unlock_and_close(self): """Close and unlock the file using the win32 primitive.""" if self._locked: try: hfile = win32file._get_osfhandle(self._fh.fileno()) win32file.UnlockFileEx(hfile, 0, -0x10000, pywintypes.OVERLAPPED()) except pywintypes.error as e: if e[0] != _Win32Opener.FILE_ALREADY_UNLOCKED_ERROR: raise self._locked = False if self._fh: self._fh.close()
def unlock_and_close(self): """Close and unlock the file using the win32 primitive.""" if self._locked: try: hfile = win32file._get_osfhandle(self._fh.fileno()) win32file.UnlockFileEx(hfile, 0, -0x10000, pywintypes.OVERLAPPED()) except pywintypes.error, e: if e[0] != _Win32Opener.FILE_ALREADY_UNLOCKED_ERROR: raise self._locked = False if self._fh: self._fh.close()
def lock(file, flags): hfile = win32file._get_osfhandle(file.fileno()) win32file.LockFileEx(hfile, flags, 0, 0x7fff0000, __overlapped)
def unlock(file): hfile = win32file._get_osfhandle(file.fileno()) win32file.UnlockFileEx(hfile, 0, 0x7fff0000, __overlapped)
def lock(file, flags): hfile = win32file._get_osfhandle(_getfd(file)) try: win32file.LockFileEx(hfile, flags, 0, -0x10000, __overlapped) except pywintypes.error, exc_value: # error: (33, 'LockFileEx', 'The process cannot access the file because another process has locked a portion of the file.') if exc_value[0] == 33: raise LockException(LockException.LOCK_FAILED, exc_value[2]) else: # Q: Are there exceptions/codes we should be dealing with here? raise
def unlock(file): hfile = win32file._get_osfhandle(_getfd(file)) try: win32file.UnlockFileEx(hfile, 0, -0x10000, __overlapped) except pywintypes.error, exc_value: if exc_value[0] == 158: # error: (158, 'UnlockFileEx', 'The segment is already unlocked.') # To match the 'posix' implementation, silently ignore this error pass else: # Q: Are there exceptions/codes we should be dealing with here? raise
def FileObject2File(fileObject): """Get the win32 file handle from a C stdio file object""" return win32file._get_osfhandle(fileObject.fileno())
def lock(file, flags): hfile = win32file._get_osfhandle(fd(file)) win32file.LockFileEx(hfile, flags, 0, -0x10000, __overlapped)
def unlock(file): hfile = win32file._get_osfhandle(fd(file)) win32file.UnlockFileEx(hfile, 0, -0x10000, __overlapped)
def lock(file, flags): hfile = win32file._get_osfhandle(file.fileno()) try: win32file.LockFileEx(hfile, flags, 0, nNumberOfBytesToLockHigh, __overlapped) except pywintypes.error as exc_value: # error: (33, 'LockFileEx', 'The process cannot access the file because another # process has locked a portion of the file.') if exc_value.winerror == 33: raise LockException(str(exc_value)) else: # Q: Are there exceptions/codes we should be dealing with here? raise
def unlock(file): hfile = win32file._get_osfhandle(file.fileno()) try: win32file.UnlockFileEx(hfile, 0, nNumberOfBytesToLockHigh, __overlapped) except pywintypes.error as exc_value: if exc_value.winerror == 158: # error: (158, 'UnlockFileEx', 'The segment is already unlocked.') # To match the 'posix' implementation, silently ignore this error pass else: # Q: Are there exceptions/codes we should be dealing with here? raise
def open_and_lock(self, timeout, delay): """Open the file and lock it. Args: timeout: float, How long to try to lock for. delay: float, How long to wait between retries Raises: AlreadyLockedException: if the lock is already acquired. IOError: if the open fails. CredentialsFileSymbolicLinkError: if the file is a symbolic link. """ if self._locked: raise AlreadyLockedException('File %s is already locked' % self._filename) start_time = time.time() validate_file(self._filename) try: self._fh = open(self._filename, self._mode) except IOError as e: # If we can't access with _mode, try _fallback_mode # and don't lock. if e.errno == errno.EACCES: self._fh = open(self._filename, self._fallback_mode) return # We opened in _mode, try to lock the file. while True: try: hfile = win32file._get_osfhandle(self._fh.fileno()) win32file.LockFileEx( hfile, (win32con.LOCKFILE_FAIL_IMMEDIATELY | win32con.LOCKFILE_EXCLUSIVE_LOCK), 0, -0x10000, pywintypes.OVERLAPPED()) self._locked = True return except pywintypes.error as e: if timeout == 0: raise # If the error is not that the file is already # in use, raise. if e[0] != _Win32Opener.FILE_IN_USE_ERROR: raise # We could not acquire the lock. Try again. if (time.time() - start_time) >= timeout: logger.warn('Could not lock %s in %s seconds' % ( self._filename, timeout)) if self._fh: self._fh.close() self._fh = open(self._filename, self._fallback_mode) return time.sleep(delay)
def open_and_lock(self, timeout, delay): """Open the file and lock it. Args: timeout: float, How long to try to lock for. delay: float, How long to wait between retries Raises: AlreadyLockedException: if the lock is already acquired. IOError: if the open fails. CredentialsFileSymbolicLinkError if the file is a symbolic link. """ if self._locked: raise AlreadyLockedException('File %s is already locked' % self._filename) start_time = time.time() validate_file(self._filename) try: self._fh = open(self._filename, self._mode) except IOError, e: # If we can't access with _mode, try _fallback_mode and don't lock. if e.errno == errno.EACCES: self._fh = open(self._filename, self._fallback_mode) return # We opened in _mode, try to lock the file. while True: try: hfile = win32file._get_osfhandle(self._fh.fileno()) win32file.LockFileEx( hfile, (win32con.LOCKFILE_FAIL_IMMEDIATELY| win32con.LOCKFILE_EXCLUSIVE_LOCK), 0, -0x10000, pywintypes.OVERLAPPED()) self._locked = True return except pywintypes.error, e: if timeout == 0: raise e # If the error is not that the file is already in use, raise. if e[0] != _Win32Opener.FILE_IN_USE_ERROR: raise # We could not acquire the lock. Try again. if (time.time() - start_time) >= timeout: logger.warn('Could not lock %s in %s seconds' % ( self._filename, timeout)) if self._fh: self._fh.close() self._fh = open(self._filename, self._fallback_mode) return time.sleep(delay)
def open_and_lock(self, timeout, delay): """Open the file and lock it. Args: timeout: float, How long to try to lock for. delay: float, How long to wait between retries Raises: AlreadyLockedException: if the lock is already acquired. IOError: if the open fails. CredentialsFileSymbolicLinkError if the file is a symbolic link. """ if self._locked: raise AlreadyLockedException('File %s is already locked' % self._filename) start_time = time.time() validate_file(self._filename) try: self._fh = open(self._filename, self._mode) except IOError as e: # If we can't access with _mode, try _fallback_mode and don't lock. if e.errno == errno.EACCES: self._fh = open(self._filename, self._fallback_mode) return # We opened in _mode, try to lock the file. while True: try: hfile = win32file._get_osfhandle(self._fh.fileno()) win32file.LockFileEx( hfile, (win32con.LOCKFILE_FAIL_IMMEDIATELY| win32con.LOCKFILE_EXCLUSIVE_LOCK), 0, -0x10000, pywintypes.OVERLAPPED()) self._locked = True return except pywintypes.error as e: if timeout == 0: raise e # If the error is not that the file is already in use, raise. if e[0] != _Win32Opener.FILE_IN_USE_ERROR: raise # We could not acquire the lock. Try again. if (time.time() - start_time) >= timeout: logger.warn('Could not lock %s in %s seconds' % ( self._filename, timeout)) if self._fh: self._fh.close() self._fh = open(self._filename, self._fallback_mode) return time.sleep(delay)
def open_and_lock(self, timeout, delay): """Open the file and lock it. Args: timeout: float, How long to try to lock for. delay: float, How long to wait between retries Raises: AlreadyLockedException: if the lock is already acquired. IOError: if the open fails. CredentialsFileSymbolicLinkError: if the file is a symbolic link. """ if self._locked: raise AlreadyLockedException( 'File {0} is already locked'.format(self._filename)) start_time = time.time() validate_file(self._filename) try: self._fh = open(self._filename, self._mode) except IOError as e: # If we can't access with _mode, try _fallback_mode # and don't lock. if e.errno == errno.EACCES: self._fh = open(self._filename, self._fallback_mode) return # We opened in _mode, try to lock the file. while True: try: hfile = win32file._get_osfhandle(self._fh.fileno()) win32file.LockFileEx( hfile, (win32con.LOCKFILE_FAIL_IMMEDIATELY | win32con.LOCKFILE_EXCLUSIVE_LOCK), 0, -0x10000, pywintypes.OVERLAPPED()) self._locked = True return except pywintypes.error as e: if timeout == 0: raise # If the error is not that the file is already # in use, raise. if e[0] != _Win32Opener.FILE_IN_USE_ERROR: raise # We could not acquire the lock. Try again. if (time.time() - start_time) >= timeout: logger.warn('Could not lock %s in %s seconds', self._filename, timeout) if self._fh: self._fh.close() self._fh = open(self._filename, self._fallback_mode) return time.sleep(delay)
