eshangj/stackoverflow_q_and_a_sample huggingface dataset:
Question: Use AsyncSession in celery tasks I use fastapi and sqlalchemy, I must create celery task, that will go to the database and check does any objects of my Event (table) has end_time < datetime.now() There is my code: @asynccontextmanager async def scoped_session(): scoped_factory = async_scoped_session( async_session, scopefunc=asyncio.current_task() ) try: async with scoped_factory() as s: yield s finally: await scoped_factory().remove() async def logic(): async with scoped_session() as session: stmt = select(event.models.Event).where( event.models.Event.end_time <= datetime.now() ) results = await session.execute(stmt) for res in results.fetchall(): print(res.is_event_done) @celery.task(name='is_event_done', bind=True, ignore_result=True) def is_event_done(self): asyncio.run(logic())  here is my async_session engine = create_async_engine(settings.db_url, echo=True) async_session = sessionmaker( engine, class_=AsyncSession, expire_on_commit=False )  so I got \'_asyncio.Task\' object is not callable

Answer: I just do like this async def update_event() -> None: async with async_session() as session: stmt = update(event.models.Event).where( event.models.Event.end_time <= datetime.now(), event.models.Event.is_active is True ).values(is_done=True) await session.execute(stmt) @celery.task(name='is_event_done', bind=True, ignore_result=True) def is_event_done(self) -> None: loop.run_until_complete(update_event())  and now its work fine, I hope that its a good solution If there is anything wrong let me know, thanks!
Question: I'm trying to build a Python package from Rust using PyO3. Right now I'm stuck trying to return enums Rust type to Python. I have a simple enum like so: pub enum Lang { Deu, Eng, Fra }  And in lib.rs #[pyfunction] fn detect_language(text: &str) -> PyResult<????> { // Do some stuff .... res:Lang = Do_some_stuff(text) Ok(res) } #[pymodule] fn pymylib(_py: Python, m: &PyModule) -> PyResult<()> { m.add_function(wrap_pyfunction!(detect_language, m)?)?; Ok(()) }  In Python code from pymylib import detect_language res=detect_language('Ceci est un test') print(res) # Lang:Fra ??? 

Answer: One approach would to use #[pyclass] attribute to make the Python class from Rust Enum. Also make sure to export this class from Rust code, so that you can do the comparison on python layer. ie, use pyo3::prelude::*; #[pyclass] pub enum Lang { Deu, Eng, Fra } #[pyfunction] fn detect_language(text: &str) -> PyResult<Lang> { // Write your actual code here // But for testing purpose let's return `Deu` varient. Ok(Lang::Deu) } /// A Python module implemented in Rust. #[pymodule] fn pymylib(_py: Python, m: &PyModule) -> PyResult<()> { m.add_function(wrap_pyfunction!(detect_language, m)?)?; m.add_class::<Lang>()?; Ok(()) }  Now in Python interpreter you can do, >>> from pymylib import Lang >>> import pymylib >>> >>> pymylib.detect_language("1") Lang.Deu >>> pymylib.detect_language("1") == Lang.Deu True  See also  Support exporting Rust enums to Python
Question: when running the below file , I am getting error "ImportError: cannot import name 'get_object_size' from 'bson' (C:\Users\Dell\AppData\Local\Programs\Python\Python310\lib\site-packages\bson_init.py)" code: `from flask import Flask, request, jsonify from flask_pymongo import PyMongo # from bson.objectid import ObjectId app = Flask(__name__) app.config['MONGO_URI'] = connectionstring mongo = PyMongo(app) # Create a new to-do item @app.route('/api/todo', methods=['POST']) def create_todo(): # data = request.json # Create a new to-do item in the database new_todo = { # 'task': data['task'], # 'due_date': data['due_date'], # 'completed': False "task": "study", "date": 18, "completed": False } result = mongo.db.todos.insert_one(new_todo) print(result,"has been created")`  I am creating a todo list using flask environment.

Answer: uninstall both pymongo and bson and install just pymongo, pymongo automatically installs bson pip uninstall pymongo  pip uninstall bson  pip install pymongo
Question: Say I have: In [1]: df = pl.DataFrame({'a': [[1,2], [3,4]]}) In [2]: df Out[2]: shape: (2, 1) ┌───────────┐ │ a │ │ --- │ │ list[i64] │ ╞═══════════╡ │ [1, 2] │ │ [3, 4] │ └───────────┘  I know that all elements of 'a' are lists of the same length. I can do: In [10]: df.select(pl.col('a').list.get(i).alias(f'a_{i}') for i in range(2)) Out[10]: shape: (2, 2) ┌─────┬─────┐ │ a_0 ┆ a_1 │ │ --- ┆ --- │ │ i64 ┆ i64 │ ╞═════╪═════╡ │ 1 ┆ 2 │ │ 3 ┆ 4 │ └─────┴─────┘  but this involved hard-coding 2. Is there a way to do this without hard-coding the 2? I may not know in advance how many elements there in the lists (I just know that they all have the same number of elements)

Answer: You can convert the list to a struct and .unnest() df.with_columns(pl.col("a").list.to_struct()).unnest("a")  shape: (2, 2) ┌─────────┬─────────┐ │ field_0 ┆ field_1 │ │ --- ┆ --- │ │ i64 ┆ i64 │ ╞═════════╪═════════╡ │ 1 ┆ 2 │ │ 3 ┆ 4 │ └─────────┴─────────┘  Warning: If your lists are not the same length, you must set n_field_strategy to max_width. .list.to_struct("max_width")  By default, it uses the length of the first list found. This would result in truncated data if you had longer lists later in your data.
Question: I would like to know how to enable the highlighting of function call in VS Code with python. See the following example where function call is blank, as other part of the code: 

Answer: See the theming docs and the Developer: Inspect Editor Tokens and Scopes command in the command palette. Ex. Semantic highlighting customization route (requires a language extension that provides semantic highlighting support for Python) (highlights both function definitions and calls): "editor.semanticTokenColorCustomizations": { "[Theme Name Goes Here]": { // remove this wrapper to apply to all themes "rules": { "function:python": { // remove ":python" to apply to all languages "foreground": "#FF0000", // TODO // "fontStyle": "" // optional } } } }  ^the :python part will make the colour customization only apply to functions for the Python language. If you want the call for function declarations to be different, then write another similar rule, but use the declaration modifier. Ex. function.declaration:python.  Token colour customization route (works without any extensions, since VS Code bundles TextMate grammar support for Python) (may require disabling extensions that provide semantic highlighting support for Python): "editor.tokenColorCustomizations": { "[Theme Name Goes Here]": { // remove this wrapper to apply to all themes "textMateRules": [ { "scope": "meta.function-call.python", "settings": { "foreground": "#FF0000", // TODO // "fontStyle": "" // optional } } ] } },  See also the scopes support.function.builtin.python, meta.function-call.generic.python, and meta.member.access.python, meta.function-call.python.
Question: I work with the OpenAI API. I have extracted slides text from a PowerPoint presentation, and written a prompt for each slide. Now, I want to make asynchronous API calls, so that all the slides are processed at the same time. this is the code from the async main function: for prompt in prompted_slides_text: task = asyncio.create_task(api_manager.generate_answer(prompt)) tasks.append(task) results = await asyncio.gather(*tasks)  and this is generate_answer function: @staticmethod async def generate_answer(prompt): """ Send a prompt to OpenAI API and get the answer. :param prompt: the prompt to send. :return: the answer. """ completion = await openai.ChatCompletion.create( model="gpt-3.5-turbo", messages=[{"role": "user", "content": prompt}] ) return completion.choices[0].message.content  the problem is:  object OpenAIObject can't be used in 'await' expression  and I don't know how to await for the response in generate_answer function Would appreciate any help!

Answer: For those landing here, the error here was probably the instantiation of the object. It has to be: client = AsyncOpenAI(api_key=api_key)  Then you can use:  response = await client.chat.completions.create( model="gpt-4", messages=custom_prompt, temperature=0.9 )
Question: I want to catch and then re-throw warnings from my Python code, similarly to try/except clause. My purpose is to catch the warning and then re-throw it using my logger. The warnings are issued from whatever packages I'm using, I would like something that is totally generic, exactly like the try/except clause. How can I do that in Python >= v3.8?

Answer: There are at least two ways to tackle this:  Record the warnings and replay them Make warnings to behave like exceptions and break the control flow  Option 1: Record warnings and replay them You could use warnings.catch_warnings with record=True to record the Warning objects. This way all your application code will get executed regardless of any warnings. import warnings import logging logger = logging.getLogger(__name__) class NotImportantWarning(UserWarning): ... def do_something(): warnings.warn("doing something", UserWarning) warnings.warn("This is not important", NotImportantWarning) # Executing our main function and recording warnings with warnings.catch_warnings(record=True) as recorded_warnings: do_something() print("done") # Handling warnings. You may log it here. for w in recorded_warnings: if isinstance(w.message, NotImportantWarning): continue # re-issue warning warnings.warn_explicit( message=w.message, category=w.category, filename=w.filename, lineno=w.lineno, source=w.source, ) # or use a logger logger.warning(str(w.message))  This will print then something like done <ipython-input-6-cc898056d044>:12: UserWarning: doing something warnings.warn("doing something", UserWarning) doing something  Notes  Using the lower level warnings.warn_explicit to re-issue the warnings. The denefit over warnings.warn is that with the warn_explicit it is possible to retain the original filename and line number information.  Option 2: Make warnings to behave like exceptions If you really want to break the control flow right at the warning, like with exceptions, it is possible with warnings.filterwarnings("error"): import warnings warnings.filterwarnings('error') try: do_something() except UserWarning as w: print(w)  Note: The code execution stops at the first warning. No code inside do_something after that warning will ever get executed.
Question: I would like to be able to accept only the next word of a github Copilot suggestion instead of the full suggestion. This is possible with VS Code as documented here. Is there a way to do this in PyCharm too?

Answer: This feature has been added with a recent update for the GitHub Copilot extension. One may now hit Ctrl + Right to accept the next word only, and in a multiline suggestion also Ctrl + Alt + Right to accept the next line only.
Question: I have a mkdocs project that resembles the following: project ├─mkdocs.yml ├─docs │ ├─home.md │ ├─chapter1.md │ ├─static ├─file.ext ├─image.png  I am trying to find a way to "attach" file1.ext to the build, for instance as a link in chapter1.md. Any suggestions how to achieve that? Detail: I want the file to be downloadable on click.

Answer: In mkdocs, to get the file to be downloadable on click using markdown, first you need to add this to your mkdocs.yml file : markdown_extensions: - attr_list  and then in your chapter1.md you can add download attribute to your link ... like so : [file.ext](../static/file.ext){:download}  Heck you can even name the downloaded file: [file.ext](../static/file.ext){:download="awesome-file"}   Explanation MkDocs converts Markdown to HTML using Python-Markdown which offers a flexible extension mechanism, which makes it possible to change and/or extend the behavior of the parser without having to edit the actual source files. The markdown_extensions in mkdocs.yml setting specifies the Python-Markdown extensions for MkDocs. adding attr_list entry point enables the Python-Markdown attribute lists extension, which adds support for HTML-style attributes using curly brackets {} and a CSS-like syntax. example: let's say we want to open a link in new tab, we can achieve this like so : [Google](https://www.google.com){:target="_blank"}
Question: I want to use langchain for my project. so I installed it using following command : pip install langchain but While importing "langchain" I am facing following Error: File /usr/lib/python3.8/typing.py:774, in _GenericAlias.__subclasscheck__(self, cls) 772 if self._special: 773 if not isinstance(cls, _GenericAlias): --> 774 return issubclass(cls, self.__origin__) 775 if cls._special: 776 return issubclass(cls.__origin__, self.__origin__) TypeError: issubclass() arg 1 must be a class  Any one who can solve this error ?

Answer: typing-inspect==0.8.0 typing_extensions==4.5.0
Question: I'm trying to write a middleware for a FastAPI project that manipulates the request headers and / or query parameters in some special cases. I've managed to capture and modify the request object in the middleware, but it seems that even if I modify the request object that is passed to the middleware, the function that serves the endpoint receives the original, unmodified request. Here is a simplified version of my implementation: from fastapi import FastAPI, Request from starlette.datastructures import MutableHeaders, QueryParams from starlette.middleware.base import BaseHTTPMiddleware class TestMiddleware(BaseHTTPMiddleware): def __init__(self, app: FastAPI): super().__init__(app) def get_modified_query_params(request: Request) -> QueryParams: pass ## Create and return new query params async def dispatch( self, request: Request, call_next, *args, **kwargs ) -> None: # Check and manipulate the X-DEVICE-TOKEN if required header_key = "X-DEVICE-INFo" new_header_value = "new device info" new_header = MutableHeaders(request._headers) new_header[header_key] = new_header_value request._headers = new_header request._query_params = self.get_modified_query_params(request) print("modified headers =>", request.headers) print("modified params =>", request.query_params) return await call_next(request)  Even though I see the updated values in the print statements above, when I try to print request object in the function that serves the endpoint, I see original values of the request. What am I missing?

Answer: To update or modify the request headers within a middleware, you would have to update request.scope['headers'], as described in this answer. In that way, you could add new custom headers, as well as modify existing ones. In a similar way, by updating request.scope['query_string'], you could modify existing, as well as add new, query parameters. A working example is given below. Working Example from fastapi import FastAPI, Request from urllib.parse import urlencode app = FastAPI() @app.middleware('http') async def some_middleware(request: Request, call_next): # update request headers headers = dict(request.scope['headers']) headers[b'custom-header'] = b'my custom header' request.scope['headers'] = [(k, v) for k, v in headers.items()] # update request query parameters q_params = dict(request.query_params) q_params['custom-q-param'] = 'my custom query param' request.scope['query_string'] = urlencode(q_params).encode('utf-8') return await call_next(request) @app.get('/') async def main(request: Request): return {'headers': request.headers, 'q_params': request.query_params}
Question: I was creating a voice assistant project but I am having problem with the line with with command in it. The code I've written is this import speech_recognition as sr import win32com.client speaker = win32com.client.Dispatch("SAPI.SpVoice") def say(text): speaker.Speak(f"{text}") def takeCommand(): r = sr.Recognizer() with sr.Microphone as source: r.pause_threshold = 1 audio = r.listen(source) query = r.recognize_google(audio, language="en-in") print(f"User said: {query}") return query if __name__ == "__main__": print("VS Code") say("Hello I am Jarvis A.I.") while 1: print("listening...") text = takeCommand() say(text)  And the error it always gets is this VS Code listening... Traceback (most recent call last): File "f:\Jarvis AI\main.py", line 23, in <module> text = takeCommand() ^^^^^^^^^^^^^ File "f:\Jarvis AI\main.py", line 11, in takeCommand with sr.Microphone as source: TypeError: 'type' object does not support the context manager protocol  I've installed packages in my system like pywin32, pyaudio and speechrecognition but now I don't know what to do and how to proceed.

Answer: try changing with sr.Microphone as source: to with sr.Microphone() as source:
Question: How is it possible, that running the same Python program twice with the exact same seeds and static data input produces different results? Calling the below function in a Jupyter Notebook yields the same results, however, when I restart the kernel, the results are different. The same applies when I run the code from the command line as a Python script. Is there anything else people do to make sure their code is reproducible? All resources I found talk about setting seeds. The randomness is introduced by ShapRFECV. This code runs on a CPU only. MWE (In this code I generate a dataset and eliminate features using ShapRFECV, if that's important): import os, random import numpy as np import pandas as pd from probatus.feature_elimination import ShapRFECV from sklearn.ensemble import RandomForestClassifier from sklearn.datasets import make_classification global_seed = 1234 os.environ['PYTHONHASHSEED'] = str(global_seed) np.random.seed(global_seed) random.seed(global_seed) feature_names = ['f1', 'f2', 'f3_static', 'f4', 'f5', 'f6', 'f7', 'f8', 'f9', 'f10', 'f11', 'f12', 'f13', 'f14', 'f15', 'f16', 'f17', 'f18', 'f19', 'f20'] # Code from tutorial on probatus documentation X, y = make_classification(n_samples=100, class_sep=0.05, n_informative=6, n_features=20, random_state=0, n_redundant=10, n_clusters_per_class=1) X = pd.DataFrame(X, columns=feature_names) def shap_feature_selection(X, y, seed: int) -> list[str]: random_forest = RandomForestClassifier(random_state=seed, n_estimators=70, max_features='log2', criterion='entropy', class_weight='balanced') # Set to run on one thread only shap_elimination = ShapRFECV(clf=random_forest, step=0.2, cv=5, scoring='f1_macro', n_jobs=1, random_state=seed) report = shap_elimination.fit_compute(X, y, check_additivity=True, seed=seed) # Return the set of features with the best validation accuracy return report.iloc[[report['val_metric_mean'].idxmax() - 1]]['features_set'].to_list()[0]  Results: # Results from the first run shap_feature_selection(X, y, 0) >>> ['f17', 'f15', 'f18', 'f8', 'f12', 'f1', 'f13'] # Running again in same session shap_feature_selection(X, y, 0) >>> ['f17', 'f15', 'f18', 'f8', 'f12', 'f1', 'f13'] # Restarting the kernel and running the exact same command shap_feature_selection(X, y, 0) >>> ['f8', 'f1', 'f17', 'f6', 'f18', 'f20', 'f12', 'f15', 'f7', 'f13', 'f11']  Details:  Ubuntu 22.04 Python 3.9.12 Numpy 1.22.0 Sklearn 1.1.1 

Answer: This has now been fixed in probatus (the issue was a bug, apparently connected to the pandas implementation they were using, see here). For me, everything works as expecting when using the probatus' latest code version (not the package).
Question: I am using VSCode with ArcGIS Pro 3.0 in a virtual environment. Until yesterday, everything worked just fine. After updating to Pro 3.0, I was still able to use open a script and then have it run in the terminal window. Previously, I was able to select a line from the script, run it, and then it would open the correct interpreter. However, now I am unable to do so and cannot troubleshoot why this is happening. I have added the correct path to the ArcGIS Pro python executable in the interpreter path, but the terminal opens to another python executable. Any advice would be greatly appreciated as to how I can run specific python executable that I want to run. UPDATE: I can open VSCode using code from my anaconda installation, but still am having trouble running python interactively in the terminal. Previously, I used to be able to do this (e.g. test indented code cells), but this doesn't seem to be functioning anymore.

Answer: I was typing python instead of Python. python was engaging python that was in PYTHONPATH.
Question: In my Power BI dashboard, I created a Python Script that accesses an API and generates a Pandas data frame. It works fine, but how can I edit the Python code? I thought it would be something simple, but I can't really find how to find it in the interface. If I send the .pbix file to someone, they will receive an alert that a Python script is executing and display the code nicely formatted. I can find the code if I go to "Model Exhibition -> Edit query -> Advanced Editor" (I'm translating the options from another language, they can be somewhat different). I is a M language code and the Python script is displayed as a long line as the image below:  I believe it is possible to open a text box to edit the Python script, but can't really find it.

Answer: In Power Query you should be able to click on the ribbon to insert Python code as below.  If the script is existing, then click the little cog icon to the right of the step in APPLIED STEPS as below:
Question: I'm trying to get the price of the game from the epic games store, but I get a 403 error import requests from bs4 import BeautifulSoup url = "https://store.epicgames.com/ru/p/cities-skylines" response = requests.get(url) if response.status_code == 200: soup = BeautifulSoup(response.text, 'html.parser') price_element = soup.find(class_='css-119zqif') if price_element is not None: price = price_element.get_text(strip=True) print("Price Cities: Skylines:", price) else: print("Error") else: print("Error:", response.status_code)  I tried to access the game page directly, and take the price information from the class in html, but I get the error "Request error: 403"

Answer: You get to the cloudflare page dedicated to fighting robots. To get around this limitation, you need to imitate a real person. To do this, you can use the following library or similar. Here is an example of working code. Don't forget to pip install cloudscraper. from bs4 import BeautifulSoup import cloudscraper url = "https://store.epicgames.com/ru/p/cities-skylines" scraper = cloudscraper.create_scraper() response = scraper.get(url) if response.status_code == 200: soup = BeautifulSoup(response.text, 'html.parser') price_element = soup.find('div', 'css-169q7x3').find(class_='css-119zqif') print(price_element) if price_element is not None: price = price_element.get_text(strip=True) print("Price Cities: Skylines:", price) else: print("Error") else: print("Error:", response.status_code)  result: Price Cities: Skylines: 330 ₽  One remark. This library has free limitations and sometimes an error may occur: cloudscraper.exceptions.CloudflareChallengeError: Detected a Cloudflare version 2 Captcha challenge, This feature is not available in the opensource (free) version. Ignore it and get the result. But you can always rewrite your code using this API. If you sign up, they will give you a token for 10,000 requests per month. If this option does not suit you, look in the direction of selenium.
Question: I'm training a Keras model and saving it for later use using pickle. When I unpickle I get this error: AttributeError: 'Adam' object has no attribute 'build' Here's the code: from tensorflow.keras.models import Sequential from tensorflow.keras.layers import Dense import pickle model = Sequential() model.add(Dense(32, activation='relu', input_shape=(3,))) model.add(Dense(1, activation='linear')) # Use linear activation for regression model.compile(loss='mean_squared_error', optimizer='adam') pickle.dump(model, open("m.pkl", 'wb')) loadedModel = pickle.load(open("m.pkl", 'rb'))  I get this error with TensorFlow 2.11.x and 2.13.0-rc0 on MacOS M1

Answer: Instead of pickling, you should save the model using h5. This solves the issue: from keras.models import load_model model.save('m.h5') loadedModel = load_model('m.h5')