Coldchain9's questions

我有一些带有类似代码的脚本：

AND EQUAL_NULL(b.DATE_FIELD, a.DATE_FIELD)

我想动态替换，因为b.DATE_FIELD并且a.DATE_FIELD可以有不同的字段。

我认为我需要将括号内的每个字段捕获为它们自己的捕获组，并将它们写入新的等效项。基本上，我认为介于(和之间的内容,将作为第 1 组，介于,和之间的内容)将作为第 2 组。然后我可以将它们写入下面显示的新替换模式。

期望的最终结果：

AND COALESCE(b.DATE_FIELD, 'NA') = COALESCE(a.DATE_FIELD, 'NA')

我的尝试：

 echo "AND EQUAL_NULL(b.DATE_LOCAL, a.DATE_LOCAL)" | sed -n "s/AND EQUAL_NULL(\([^,]*\)\([^)]*\).*/COALESCE(\1\, \'NA\'\) = COALESCE(\2\, \'NA\'\)/p"

返回我想要的大部分内容： COALESCE(b.DATE_LOCAL, 'NA') = COALESCE(, a.DATE_LOCAL, 'NA')

我不太清楚如何获取b.DATE_FIELDas Group\1和a.DATE_FIELDAS Group\2

更新的代码更接近解决方案，但仍缺少一部分

在 Linux 上使用 GNU Sed 4.9 和 zsh 进行了测试

我有一个场景，其中有带有静态方法的对象。它们都是使用外部def build_hello()作为类变量构建的。

def build_hello(name: str):
    @staticmethod
    def hello_fn():
        return "hello my name is "

    # Assign an attribute to the staticmethod so it can be used across all classes
    hello_fn.first_name = name
    print(hello_fn() + hello_fn.first_name) # This works
    return hello_fn

class World:
    hello_fn = build_hello("bob")

# Error, function object has no attribute "first_name"
World.hello_fn.first_name

这是怎么回事？我能够在函数调用hello_fn()中访问属性build_hello()。但是当将其添加到我的对象时，该属性不再列出。

如果我调用dir()静态方法，我也看不到它：

dir(World.hello_fn)
['__annotations__',
 '__builtins__',
 '__call__',
 '__class__',
 '__closure__',
 '__code__',
 '__defaults__',
 '__delattr__',
 '__dict__',
 '__dir__',
 '__doc__',
 '__eq__',
 '__format__',
 '__ge__',
 '__get__',
 '__getattribute__',
 '__getstate__',
 '__globals__',
 '__gt__',
 '__hash__',
 '__init__',
 '__init_subclass__',
 '__kwdefaults__',
 '__le__',
 '__lt__',
 '__module__',
 '__name__',
 '__ne__',
 '__new__',
 '__qualname__',
 '__reduce__',
 '__reduce_ex__',
 '__repr__',
 '__setattr__',
 '__sizeof__',
 '__str__',
 '__subclasshook__',
 '__type_params__']

我正在向外部 API 发出请求，获取响应并将其写入文件。一切正常/运行正常，但我收到“任务已被销毁但处于待处理状态！”警告，我想清理一下。

我模拟了以下过程。我从项目源（例如列表）接收项目，并在收到项目时将它们放入队列，发出列表末尾的信号，但会放入一个标记None值。

继续write_q从该队列接收项目，直到收到哨兵然后突围。

下面的代码将显示我的write_task()操作在完成之前被取消。处理此问题的正确设计是什么？

import asyncio
import aiofiles
import aiocsv
import json

async def task(l: list,  write_q: asyncio.Queue) -> None:

    # Read tasks from source of data
    for i in l:
        # Put a request task into the queue
        req: dict = {
            "headers": {"Accept": "application/json"},
            "url": "https://httpbin.org/post",
            "data": i
        }
        await write_q.put(req)

    # Sentinel value to signal we are done receiving from source
    await write_q.put(None)

async def write_task(write_q: asyncio.Queue) -> None:

    headers: bool = True
    while True:
        async with aiofiles.open("file.csv", mode="a+", newline='') as f:
            w = aiocsv.AsyncWriter(f)
            # Get data out of the queue to write it
            data = await write_q.get()
            if not data:
                write_q.task_done()
                await f.flush()
                break

            if headers:
                await w.writerow([
                    "status",
                    "data",
                ])
                headers = False

            # Write the data from the response
            await w.writerow([
                "200",
                json.dumps(data)
            ])
            write_q.task_done()

async def main() -> None:

    # Create fake data to POST
    items: list[str] = [["hello", "world"], ["asyncio", "test"]] * 5 

    # Queues for orchestrating 
    write_q = asyncio.Queue()
    producer = asyncio.create_task(
        task(items, write_q)
    )
    consumer = asyncio.create_task(
        write_task(write_q)
    )

    errors = await asyncio.gather(producer, return_exceptions=True)
    print(f"INFO: Producer has completed! exceptions: {errors}")

    # Wait for queue to empty and cancel the consumer
    await write_q.join()
    consumer.cancel()
    print("INFO: write consumer has completed! ")
    print("INFO: Complete!")

if __name__ == "__main__":
    loop = asyncio.new_event_loop()
    loop.run_until_complete(main())

我有一个数据管道，我在其中发出返回Vec<T>.每个T都有我将用来发出另一个 API 请求的信息。

首先，我流式传输请求并在包含这些请求的通道上发送Vec<T>，然后我的目标是获取该 Vector，对其进行迭代，并生成新任务以发出下一步请求。

然而，我在这个设计中迷失了方向，目前我的代码可以运行，但没有提供任何有意义的东西。

我尝试tokio::spawn在迭代器调用之前移动第二个调用（但while let(Some(r) = post_rx.recv().awaititer() 不是 asnyc，所以我收到编译错误。

本质上，我试图找出一种可以通过通道消耗响应并产生进一步任务的方法。

下面的代码（将运行，但卡在第二个通道中）：

use futures::stream::{StreamExt, TryStreamExt};
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;

#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct Post {
    pub user_id: u16,
    pub id: u16,
    pub title: String,
    pub body: String,
}

#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct User {
    pub id: u16,
    pub name: String,
    pub username: String,
    pub email: String,
    pub address: Address,
    pub phone: String,
    pub website: String,
    pub company: Company,
}

#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct Address {
    pub street: String,
    pub suite: String,
    pub city: String,
    pub zipcode: String,
    pub geo: Geo,
}

#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct Geo {
    pub lat: String,
    pub lng: String,
}

#[derive(Deserialize, Serialize, Debug, Clone)]
#[serde(rename_all = "camelCase")]
pub struct Company {
    pub name: String,
    pub catch_phrase: String,
    pub bs: String,
}

#[tokio::main]
async fn main()  {
    let (post_tx, mut post_rx) = mpsc::channel::<Vec<Post>>(5);
    let (user_tx, mut user_rx) = mpsc::channel::<User>(5);
    let client = reqwest::Client::new();
    // create iterator that will stream async responses
    tokio::spawn(async move {
        let _ = futures::stream::iter((1..).step_by(1)).then(|i| {
            let client = client.clone();
            let url = format!("https://jsonplaceholder.typicode.com/posts?userId={i}");
            client.get(url).send()
        })
        .and_then(|resp| {
            resp.json::<Vec<Post>>()
        })
        .try_for_each_concurrent(2, |r| async {
            let post_tx_cloned = post_tx.clone();
            let _ = post_tx_cloned.send(r).await;
            Ok(())
        }).await;
    });

    // consume responses from our channel to do future things with results...
    let new_client = reqwest::Client::new();
    while let Some(r) = post_rx.recv().await {

        println!("received {} posts", r.len());

        // iterate over vec of posts to spawn other concurrent requests
        let posts = r.clone();
        let _ = posts.iter().map(|p| {
            let p = p.clone();
            let cloned_client = new_client.clone();
            let user_tx_cloned = user_tx.clone();
            // should I move this task spawn before the iter().map() call?
            tokio::spawn(async move {
                let url = format!("https://jsonplaceholder.typicode.com/users/{id}", id = p.user_id);
                let resp = cloned_client
                    .get(url)
                    .send()
                    .await
                    .unwrap()
                    .json::<User>()
                    .await;
                println!("{:?}", resp);
                user_tx_cloned.send(resp.unwrap()).await.unwrap();
            });
        });

        // quit if the response comes back empty
        if r.len() == 0 {
            break;
        }
    }

    // read from the user channel
    while let Some(user) = user_rx.recv().await {
        println!("{:?}", user);
    }
}

编辑：我更新了 main 以生成一个 for 循环（而不是iter().map()），它现在消耗第二个通道中的用户。我相信问题是因为我没有使用.map()上面的先前代码中的 。在这种情况下，使用 for 循环似乎更简单。代码现在运行，但从未完成。

编辑后的代码：

#[tokio::main]
async fn main()  {
    let (post_tx, mut post_rx) = mpsc::channel::<Vec<Post>>(5);
    let (user_tx, mut user_rx) = mpsc::channel::<User>(5);
    let client = reqwest::Client::new();
    // create iterator that will stream async responses
    tokio::spawn(async move {
        let _ = futures::stream::iter((1..).step_by(1)).then(|i| {
            let client = client.clone();
            let url = format!("https://jsonplaceholder.typicode.com/posts?userId={i}");
            client.get(url).send()
        })
        .and_then(|resp| {
            resp.json::<Vec<Post>>()
        })
        .try_for_each_concurrent(2, |r| async {
            let post_tx_cloned = post_tx.clone();
            let _ = post_tx_cloned.send(r).await;
            Ok(())
        }).await;
    });

    // consume responses from our channel to do future things with results...
    while let Some(r) = post_rx.recv().await {

        println!("received {} posts", r.len());

        // iterate over vec of posts to spawn other concurrent requests
        let posts = r.clone();
        // for i in posts.iter() {
        //     println!("post: {:?}", i);
        // }
        let new_client = reqwest::Client::new();
        let cloned_client = new_client.clone();
        let user_tx_cloned = user_tx.clone();
        tokio::spawn(async move {
            for p in posts {
                let url = format!("https://jsonplaceholder.typicode.com/users/{id}", id = p.user_id);
                let resp = cloned_client
                    .get(url)
                    .send()
                    .await
                    .unwrap()
                    .json::<User>()
                    .await;
                user_tx_cloned.send(resp.unwrap()).await.unwrap();
            }
        });

        // quit if the response comes back empty
        if r.len() == 0 {
            break;
        }
    }

    // read from the user channel
    while let Some(user) = user_rx.recv().await {
        println!("found user {}", user.id);
    }
}

我有一本包含三个（可变）项目的字典：

d = {"data": [1, 2, 3, 4], "id": 4, "name": "bob"}

我想将其分成两个字典而不复制任何数据。基本上我想将id和name键删除到它自己的字典中，但只能通过指定“我想要所有不是“数据”键的键。

我想避免复制“数据”键值项，因为它的大小可能为 15+MB，因此，如果有的话，需要提取其他项并保持“数据”键值项完好无损。

我的尝试：

# This will create a new dict for the non "data" key items. Great!
# But I want the "data" key-value pair to remain in its own dictionary without having to copy it into a new dict
non_data_dict = {key: d[key] for key in d.keys() - ['data']}

我想要的结果：

data_dict = {"data": [1,2,3,4]}
non_data_dict = {"id": 4, "name": "bob"}