标签: contextmanager

这段带有pandas 1.5.3和sqlalchemy 的 2.0.1代码不再工作，令人惊讶的是，它不会引发任何错误，代码默默地通过：

# python 3.10.6
import pandas as pd # 1.5.3
import psycopg2 # '2.9.5 (dt dec pq3 ext lo64)'
from sqlalchemy import create_engine # 2.0.1


def connector():
    return psycopg2.connect(**DB_PARAMS)

engine = create_engine('postgresql+psycopg2://', creator=connector)

with engine.connect() as connection:
    df.to_sql(
        name='my_table',
        con=connection,
        if_exists='replace',
        index=False,
    )

目前，使用 sqlalchemy，2.0.1我的表不再填充 DataFrame 内容。

而它已正确填充 sqlalchemy version 1.4.45。

编辑

显然，当我不使用上下文管理器时它会起作用：

connection = engine.connect()

res.to_sql(
    name='my_table',
    con=connection,
    if_exists='replace',
    index=False
)
Out[2]: 133 # <- wondering what …

我倾向于经常使用Python"with"语句.在我将一些文件符号链接或复制到目录后,主要用于清理目录,因为即使python脚本崩溃,任务仍然会执行.以下是我的函数的示例,可以与"with"语句一起使用.

@contextmanager
def use_symlink(orig, dest):
    os.symlink(orig, dest)
    try: 
        yield
    finally:
        os.unlink(link)

我使用这些语句的方式很快就会堆积起来.

#Off to an adventure
with use_symlink(a, b):
    with use_symlink(c, b):
        with use_symlink(d, b):
            with working_dir(dir1):
                #do something
            with working_dir(dir2):
                #do something that creates file dir2_file1, dir2_file2
                with use_symlink(dir2_file1, b):
                   with use_symlink(dir2_file2, b):
                       with working_dir(b):
                           #Do the last thing
#Home safely

是否有更好的方法来执行上述操作与强大的"with"语句相同的简易性和安全性？

我想要这样的行为:

with A() as f:
    for x in f:
        do_something(f)

这是正确的方法吗？

class A:
    def __enter__(self):
        print "Entering context"

    def __iter__(self):
        for x in ["some","list"]:
            yield x

    def __exit__(self):
        print "Deleting context"

我有一个脚本,实例化了许多子对象.当脚本(和这些对象)结束时......对象需要做一些清理(尤其是临时文件关闭和删除).

我一直在阅读如何__del__不可靠 - 但是上下文管理似乎不起作用,子对象不会持久.他们需要四处闲逛(做文件读写等事情)

一个例子:

WRITER.PY

import os

class Writer(object):
    def __init__(self, filename):
        self.filename = filename
        self.open()

    def open(self):
        self.fh = open(self.filename, "w+", 0)

    def write(self, msg):
        print >>self.fh, str(msg)

    def close(self):
        self.fh.close()
        os.remove(self.filename)

    def __enter__(self):
        print "entered"
#         self.open()


    def __exit__(self, type, value, traceback):
        print "__exit__"
        self.close()

MAIN.PY

def myfunc(filename):
    with WRITER.Writer(filename) as writeit:
        # do some stuff
        writeit.write("hallo")
        # do some more stuff
        writeit.write("more results")
        # even more stuff
        writeit.write("goodbye")

但是当我运行时myfunc(),对象一旦完成就会被垃圾收集__init__().它直接从enter到exit并且在with语句之后不执行任何任务.如果我把开放放入__init__ …

我想完成以下事情:

class my_context(object):
    def __init__(self):
        self.obj1 = Obj()
        self.obj2 = Obj()
        ...

    def __enter__(self):
        ''' initialize objects '''

    def __exit__(self, type, value, tb):
        ''' uninitialize objects '''

有许多Obj属性是需要关闭/删除/等的资源.我希望使用上下文管理器来设置它们然后摆脱它们.但是我发现我尝试时无法访问属性:

with my_context() as cont:
    cont.obj1  # doesn't work

有没有办法可以访问这些属性？

我在python中使用上下文管理器。想要从我的__exit__方法中获取一些日志。所以我的代码记录如下：

class MyContextManager:
    def __init__(self, value1, value2)
        self.value1 = value1
        self.value2 = value2

    def __enter__(self)
        # Do some other stuff
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        # Do some tear down action, process some data that is 
        # created in __enter__ and log those results
        return my_results

with MyContextManager(value1=my_value1, value2=my_value2) as manager:
     # Do some stuff

因此，如何访问__exit__with块之后（或末尾）返回的my_results 。在该__exit__方法中返回其他True甚至合法吗？

我有一个将在其中包含aiohttp.ClientSession对象的类。

通常当您使用

async with aiohttp.ClientSession() as session:  
   # some code

由于调用了会话的__aexit__方法，因此该会话将关闭。

我无法使用上下文管理器，因为我想在对象的整个生命周期中保持会话的持久性。

这有效：

import asyncio
import aiohttp

class MyAPI:
    def __init__(self):
        self.session = aiohttp.ClientSession()

    def __del__(self):
        # Close connection when this object is destroyed
        print('In __del__ now')
        asyncio.shield(self.session.__aexit__(None, None, None))



async def main():
    api = MyAPI()

asyncio.run(main())

但是，如果在某些地方引发了异常，则在完成__aexit__方法之前将关闭事件循环。我该如何克服？

堆栈跟踪：

Traceback (most recent call last):
  File "/home/ron/.PyCharm2018.3/config/scratches/async.py", line 19, in <module>
    asyncio.run(main())
  File "/usr/local/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 568, in run_until_complete
    return future.result()
  File "/home/ron/.PyCharm2018.3/config/scratches/async.py", line …

在文档中，Manager与上下文管理器（即with）一起使用，如下所示：

from multiprocessing.managers import BaseManager

class MathsClass:
    def add(self, x, y):
        return x + y
    def mul(self, x, y):
        return x * y

class MyManager(BaseManager):
    pass

MyManager.register('Maths', MathsClass)

if __name__ == '__main__':
    with MyManager() as manager:
        maths = manager.Maths()
        print(maths.add(4, 3))         # prints 7
        print(maths.mul(7, 8))         # prints 56

但是，除了名称空间之外，这样做还有什么好处？对于打开文件流，其好处是显而易见的，因为您不必手动.close()进行连接，但这对Manager有什么作用？如果您不在上下文中使用它，则必须使用什么步骤来确保所有内容均正确关闭？

简而言之，与上述类似，使用以上内容有什么好处：

manager = MyManager()
maths = manager.Maths()
print(maths.add(4, 3))         # prints 7
print(maths.mul(7, 8))         # prints 56

我无法模拟 psycopg2 数据库连接和游标，因为我已将其重构为使用游标的上下文管理器。我知道使用上下文管理器时，会调用额外的魔术方法来进行资源设置和清理（__enter__，__exit__），但即使将其添加到组合中也没有缩小问题的范围。

这是代码：

import os
import psycopg2
import psycopg2.extras

DB_HOST = os.getenv('DB_HOST')     
DB_PORT = os.getenv('DB_PORT')
DB_NAME = os.getenv('DB_NAME')
DB_USER = os.getenv('DB_USER')
DB_PASSWORD = os.getenv('DB_PASSWORD')            
CONN = psycopg2.connect(f'dbname={DB_NAME} user={DB_USER} host={DB_HOST} port={DB_PORT} password={DB_PASSWORD}')

def my_func():
    message = None
    print(CONN)  # Added to debug - this never prints out a magic mock reference

    select_sql = 'SELECT * FROM app.users WHERE name = %s LIMIT 1;'
    with CONN.cursor(cursor_factory = psycopg2.extras.DictCursor) as cursor:
        print(cursor) # Debug - no magic mock …

使用 Python 3.5.x，没有比这更高的版本。

/sf/answers/2117818601/是正确的答案，但没有提供内置于 Python 的解决方案，而是需要从头开始编写代码：

我需要一个值为“-”的字符串来表示标准输入，或者它的值是我想要读取的文本文件的路径。我想使用with运算符打开这些文件中的任何一种，而不使用条件逻辑来检查脚本中的“-”。我有一些有用的东西，但它似乎应该是内置在 Python 核心中的东西，而不需要我推出自己的上下文管理器，如下所示：

from contextlib import contextmanager

@contextmanager
def read_text_file_or_stdin(path):
    """Return a file object from stdin if path is '-', else read from path as a text file."""
    if path == '-':
        with open(0) as f:
            yield f
    else:
        with open(path, 'r') as f:
            yield f


# path = '-'  # Means read from stdin
path = '/tmp/paths'  # Means read from a text file given by this value
with read_text_file_or_stdin(path) …

我尝试使用类型提示来指定实现连接器类（在本例中为代理）时要遵循的 API。

我想指定这样的类应该是上下文管理器

我怎么做？

让我更清楚地重新表述一下：我如何定义Broker类，以便它表明它的具体实现（例如Rabbit类）必须是上下文管理器？

有没有实用的办法呢？我是否必须指定__enter__and__exit__并继承自Protocol？

继承就足够了吗ContextManager？

顺便问一下，我应该使用@runtimeor@runtime_checkable吗？（我的 VScode linter 似乎在查找 . 中的内容时遇到问题typing。我使用的是 python 3 7.5）

我知道如何使用 ABC 来做到这一点，但我想学习如何使用协议定义来做到这一点（我已经很好地使用了协议定义，但它们不是上下文管理器）。

我不知道如何使用该ContextManager类型。到目前为止我还没能从官方文档中找到好的例子。

目前我想出了

from typing import Protocol, ContextManager, runtime, Dict, List


@runtime
class Broker(ContextManager):
    """
    Basic interface to a broker.
    It must be a context manager
    """

    def publish(self, data: str) -> None:
        """
        Publish data to the topic/queue
        """
        ... …

从Python的源代码来看open,我认为open这只是一个普通的函数.

为什么我们可以像下面这样使用它？

with open('what_are_context_managers.txt', 'r') as infile:
    for line in infile:
        print('> {}'.format(line))

因为既不是实现__enter__也不是__exit__,也不使用contextlib.contextmanager装饰器.