Function NB_MODULE

example/python.cpp:312–366

NB_MODULE(boost_cobalt_example_python, m)
{
  namespace execution = asio::execution;
  m.def("__rethrow_exception", &std::rethrow_exception);

  py::class_<std::unique_ptr<asio::execution_context>>(m, "__asio__execution_context");

  // use some inlined python to get a future
  py::object locals = py::dict();
  // language=python
  PyRun_String(
      R"(def __await_impl(self):
    import asyncio
    lp = asyncio.get_event_loop()
    ft = lp.create_future()
    self.initiate(lp, ft)
    res = yield from ft
    return res)", Py_single_input, PyEval_GetGlobals(), locals.ptr());
  py::object await_impl = locals["__await_impl"];

  // language=python
  PyRun_String(
      R"(async def __aiter_impl(self):
    while not self.done:
        yield await self
)", Py_single_input, PyEval_GetGlobals(), locals.ptr());

  py::object aiter_impl = locals["__aiter_impl"];

  py::class_<py_coroutine> ct(m, "__cobalt_coroutine");
  ct.def("initiate", &py_coroutine::initiate)
    .def_prop_ro("done", &py_coroutine::done);
  setattr(ct, "__await__", await_impl);
  setattr(ct, "__aiter__", aiter_impl);

  m.def("test_generator",
        []() -> py_coroutine
        {
            BOOST_COBALT_FOR(auto v, test_generator())
              co_yield v;
            co_return py::none();
        });

  m.def("test_promise",
        []() -> py_coroutine
        {
          co_return co_await test_promise();
        });
  m.def("test_py_promise",
        [](py::object obj) -> py_coroutine
        {
          co_await await_py_coroutine(std::move(obj));
          co_return py::none();
        });
}