说明：仅供学习使用，请勿用于非法用途，若有侵权，请联系博主删除

作者：zhu6201976

博客：https://blog.csdn.net/zhu6201976

一、任何实现 __enter__() 和 __exit__() 方法的对象都可称之为上下文管理器，上下文管理器对象可以使用 with 关键字。

Python常见上下文管理器：文件、线程锁、Semaphore、socket等，其内部均已实现  __enter__() 和 __exit__() 方法。

# threading.Lock源码
class Lock:
    def __init__(self) -> None: ...
    def __enter__(self) -> bool: ...
    def __exit__(
        self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
    ) -> Optional[bool]: ...
    if sys.version_info >= (3,):
        def acquire(self, blocking: bool = ..., timeout: float = ...) -> bool: ...
    else:
        def acquire(self, blocking: bool = ...) -> bool: ...
    def release(self) -> None: ...
    def locked(self) -> bool: ...

# threading.Semaphore源码
class Semaphore:
    """This class implements semaphore objects.
    Semaphores manage a counter representing the number of release() calls minus
    the number of acquire() calls, plus an initial value. The acquire() method
    blocks if necessary until it can return without making the counter
    negative. If not given, value defaults to 1.
    """
 
    # After Tim Peters' semaphore class, but not quite the same (no maximum)
 
    def __init__(self, value=1):
        if value < 0:
            raise ValueError("semaphore initial value must be >= 0")
        self._cond = Condition(Lock())
        self._value = value
 
    def acquire(self, blocking=True, timeout=None):
        """Acquire a semaphore, decrementing the internal counter by one.
        When invoked without arguments: if the internal counter is larger than
        zero on entry, decrement it by one and return immediately. If it is zero
        on entry, block, waiting until some other thread has called release() to
        make it larger than zero. This is done with proper interlocking so that
        if multiple acquire() calls are blocked, release() will wake exactly one
        of them up. The implementation may pick one at random, so the order in
        which blocked threads are awakened should not be relied on. There is no
        return value in this case.
        When invoked with blocking set to true, do the same thing as when called
        without arguments, and return true.
        When invoked with blocking set to false, do not block. If a call without
        an argument would block, return false immediately; otherwise, do the
        same thing as when called without arguments, and return true.
        When invoked with a timeout other than None, it will block for at
        most timeout seconds.  If acquire does not complete successfully in
        that interval, return false.  Return true otherwise.
        """
        if not blocking and timeout is not None:
            raise ValueError("can't specify timeout for non-blocking acquire")
        rc = False
        endtime = None
        with self._cond:
            while self._value == 0:
                if not blocking:
                    break
                if timeout is not None:
                    if endtime is None:
                        endtime = _time() + timeout
                    else:
                        timeout = endtime - _time()
                        if timeout <= 0:
                            break
                self._cond.wait(timeout)
            else:
                self._value -= 1
                rc = True
        return rc
 
    __enter__ = acquire
 
    def release(self):
        """Release a semaphore, incrementing the internal counter by one.
        When the counter is zero on entry and another thread is waiting for it
        to become larger than zero again, wake up that thread.
        """
        with self._cond:
            self._value += 1
            self._cond.notify()
 
    def __exit__(self, t, v, tb):
        self.release()

二、模拟实现文件上下文管理器

class MyOpen(object):
    def __init__(self, file_path, mode):
        print('__init__')
        self.f = open(file_path, mode)
 
    def __enter__(self):
        print('__enter__')
        return self.f
 
    def __exit__(self, exc_type, exc_val, exc_tb):
        print('__exit__')
        self.f.close()
 
 
with MyOpen('test.txt', 'r') as f:
    content = f.content()
    print(content)

三、数据库上下文管理器

import pymysql
 
 
class Db(object):
    def __init__(self):
        print('__init__')
        self.connect = pymysql.connect(
            host='localhost',
            user='root',
            password='root',
            database='test',
            port=3306,
            charset='utf8'
        )
        self.cursor = self.connect.cursor()
 
    def __enter__(self):
        print('__enter__')
        return self
 
    def fetchall(self, sql):
        self.cursor.execute(sql)
        ret = self.cursor.fetchall()
        return ret
 
    def __exit__(self, exc_type, exc_val, exc_tb):
        print('__exit__')
        self.cursor.close()
 
 
with Db() as db:
    ret = db.fetchall('select g.id,g.name,c.name,b.name from goods as g ' +
                     'left join goods_cates as c on g.cate_id=c.id ' +
                     'left join goods_brands as b on g.brand_id=b.id')
print(ret)
 

执行结果：

四、socket上下文管理器

参考：https://wx-zhu6201976.blog.csdn.net/article/details/116715370