Movatterモバイル変換


[0]ホーム

URL:


Skip to content
DEV Community
Log in Create account

DEV Community

Cover image for How to Connect to EdgeDB in Python
Tyler Matteson
Tyler Matteson

Posted on

     

How to Connect to EdgeDB in Python

EdgeDB is 'an Object-Relation Database'. That's a fancy way of saying that it's a hybrid between tabular systems (like Postgres and MySQL) and document- style or graph-like systems (like MongoDB or Neo4j). Its feature set is really impressive, but for this article we're going to focus on one small task: connecting to the database from Python. No queries, no schema; just one thing in (hopefully) a digestible amount of detail. We're going to use a few recent Python features as well: type annotations, dataclasses and f-strings.

Our Goals:

  • Connect Synchronously
  • Test this so we can prove we're not crazy
  • Connect Asynchronously
  • Connect with an Asynchronus Pool
  • Switch back and forth between async and sync as appropriate

Setting up

For this tutorial you're going to want have Docker and Python 3.8 installed. If you're not fluent with Docker, don't worry. We're going to be running one command and then ignoring it while it runs EdgeDB in the background.

You'll want to havesome kind of virtual environment. For this experiment we'll be using Poetry, but venv, Pipenv or Dephell would work just as well.

@agritheory:~$mkdiredgedb_connect@agritheory:~$cdedgedb_connect@agritheory:~/edgedb_connect$poetry initThiscommandwill guide you through creating your pyproject.toml config.Package name[edgedb_connect]:  Version[0.1.0]:  Description[]:  Author[Tyler Matteson <tyler@agritheory.com>, n to skip]:  License[]:  Compatible Python versions[^3.8]:  Would you like to define your main dependencies interactively?(yes/no)[yes]yesYou can specify a packageinthe following forms:  - A single name(requests)  - A name and a constraint(requests ^2.23.0)  - A git url(git+https://github.com/python-poetry/poetry.git)  - A git url with a revision(git+https://github.com/python-poetry/poetry.git#develop)  - A file path(../my-package/my-package.whl)  - A directory(../my-package/)  - An url(https://example.com/packages/my-package-0.1.0.tar.gz)Searchforpackage to add(or leave blank tocontinue): edgedbFound 3 packages matching edgedbEnter package# to add, or the complete package name if it is not listed:[0] edgedb[1] edgeql-queries[2] edb> 0Enter the version constraint to require(or leave blank to use the latest version):Using version ^0.7.1foredgedbAdd a package:Would you like to define your development dependencies interactively?(yes/no)[yes] noGenerated file[tool.poetry]name="edgedb_connect"version="0.1.0"description=""authors=["Tyler Matteson <tyler@agritheory.com>"][tool.poetry.dependencies]python="^3.8"edgedb="^0.7.1"[tool.poetry.dev-dependencies][build-system]requires=["poetry>=0.12"]build-backend="poetry.masonry.api"Do you confirm generation?(yes/no)[yes]yes

Cool. Now let's get that Docker thing going. We'll only be using the 5656 port and won't be binding any data, so don't take this as instructions for running an EdgeDB docker containercorrectly. In anew terminal window:

@agritheory:~/edgedb_connect$docker run-it--rm-p 5656:5656-p 8888:8888-p 8889:8889 edgedb/edgedb

Let's make a Connection Object

Now let's create a file and write some Python.

@agritheory:~/edgedb_connect$touchconnect.py

In our newconnect.py file, let's import all of our dependencies:

from__future__importannotationsimporttypingfromdataclassesimportdataclassimportedgedb

Without going into too much detail we're going to use a Dataclass to store our connection parameters. The attribute names will match the API for connection parameters as documented in the EdgeDB Python client.

@dataclassclassEdgeDBConnection:dsn:typing.Optional[str]=Nonehost:typing.Optional[str]=Noneport:int=5656admin:typing.Optional[bool]=Falseuser:typing.Optional[str]=Nonepassword:typing.Optional[str]=Nonedatabase:typing.Optional[str]=Nonetimeout:int=60

It turns out there quite a few options available for connecting to EdgeDB. We'renot going to be using the DSN API and wewill be defaulting to connecting over a UNIX socket on the default port of 5656, which you may remember seeing in the Docker command. (The 8888 and 8889 ports are used for HTTP and GraphQL and those features are out of scope for this article.)

Let's write a class method to connect to EdgeDB.

defconnect_sync(self,connection:typing.Optional[EdgeDBConnection]=None,)->edgedb.BlockingIOConnection:returnedgedb.connect(dsn=self.dsn,host=self.host,port=self.port,admin=bool(self.admin),user=self.user,password=self.password,database=self.database,timeout=self.timeout,)

Between these type hints and the ones from the class declaration it should be pretty easy to see what's going on here. We're creating a wrapper around the connection parameters and a way to call it:edgedb.connect()

This is not TDD. It's 'testing early'.

That all looks like it should work and if you wanted to, you could import it into the repl and start interacting with EdgeDB. But we're not going to do that. We're going to be good citizens and write a test that tests this method.

To get started with the testing, we'll need to add some dependencies to our project. (We're going to add pytest's asyncio utils here preemptively).

@agritheory:~/edgedb_connect$poetry add pytest pytest-asyncio--devUsing version ^5.4.1forpytestUsing version ^0.10.0forpytest-asyncioUpdating dependenciesResolving dependencies...(0.6s)Writing lock filePackage operations: 11 installs, 0 updates, 0 removals  - Installing pyparsing(2.4.7)  - Installing six(1.14.0)  - Installing attrs(19.3.0)  - Installing more-itertools(8.2.0)  - Installing packaging(20.3)  - Installing pluggy(0.13.1)  - Installing py(1.8.1)  - Installing wcwidth(0.1.9)  - Installing pytest(5.4.1)  - Installing edgedb(0.7.1)  - Installing pytest-asyncio(0.10.0)@agritheory:~/edgedb_connect$touchtest.py

In our newly createdtest.py file let's see what we can break. First our dependencies:

importtypingimportpytestimportedgedbfromconnectimportEdgeDBConnection

The Docker image uses'edgedb' for user, password and database name. Since we want reuse these connection parameters for all of our tests, we're going to create a pytest fixture. Pytest fixtures allow you to share a variable or object between multiple tests by passing it into the test function as an argument.

@pytest.fixture(scope="module")defconnection_object()->EdgeDBConnection:returnEdgeDBConnection(dsn=None,host="localhost",port=5656,admin=False,user="edgedb",password="edgedb",database="edgedb",timeout=60,)

The hardest part of writing tests is getting started. The next hardest part is deciding what to test. In this case lets start with a simple sanity check on our fixture. If this test passes then we've confirmed that our fixture is behaving as expected.

@pytest.mark.usefixtures("connection_object")deftest_connection_object(connection_object)->None:assertconnection_object.host=="localhost"assertconnection_object.port==5656assertconnection_object.adminisFalseassertconnection_object.timeout==60assertconnection_object.user=="edgedb"assertconnection_object.password=="edgedb"assertconnection_object.database=="edgedb"

Let's run this test:

@agritheory:~/edgedb_connect$ poetry shell(.venv) @agritheory:~/edgedb_connect$ python -m pytest test.py================================================================ test session starts =================================================================platform linux -- Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1rootdir: /home/tyler/edgedb_connectplugins: asyncio-0.10.0collected 1 item                                                                                                                                     test.py .                                                                                                                                      [100%]================================================================= 1 passed in 0.03s ==================================================================

Well that's a relief. But we haven't actually connect to EdgeDB yet. Let's write a test for that.

@pytest.mark.usefixtures("connection_object")deftest_edgedb_sync_connection(connection_object)->None:sync_connection=connection_object.connect_sync()assertisinstance(sync_connection,edgedb.BlockingIOConnection)sync_connection.close()assertsync_connection.is_closed()isTrue

Since it's polite to close your database connection when you're done with it, we'll do that and assert that it is actually closed. Both theclose andis_closed methods are coming from theBlockingIOConnection class. Let's run the test.

================================================================ test session starts =================================================================platform linux -- Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1rootdir: /home/tyler/edgedb_connectplugins: asyncio-0.10.0collected 2 items                                                                                                                                    test.py ..                                                                                                                                     [100%]================================================================= 2 passed in 0.79s ==================================================================

Cool. If you never want to use the Async functionality of the EdgeDB library, go ahead and bail now, but we haven't gotten to the best part yet.

Time Warp

Let's add some async functionality to this class so we can reuse the same connection parameter boilerplate.

asyncdefconnect_async(self,connection:typing.Optional[EdgeDBConnection]=None,)->edgedb.AsyncIOConnection:returnawaitedgedb.async_connect(dsn=self.dsn,host=self.host,port=self.port,admin=bool(self.admin),user=self.user,password=self.password,database=self.database,timeout=self.timeout,)

That's barely different than the synchronous connection method! That can't be right. Let's write a test to find out.

@pytest.mark.usefixtures("connection_object")@pytest.mark.asyncioasyncdeftest_edgedb_async_connections(connection_object)->typing.NoReturn:async_connection=awaitconnection_object.connect_async()assertisinstance(async_connection,edgedb.AsyncIOConnection)awaitasync_connection.aclose()assertasync_connection.is_closed()isTrue

Test results:

(.venv) @agritheory:~/edgedb_connect$ python -m pytest test.py================================================================ test session starts =================================================================platform linux -- Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1rootdir: /home/tyler/edgedb_connectplugins: asyncio-0.10.0collected 3 items                                                                                                                                    test.py ...                                                                                                                                    [100%]================================================================== warnings summary ==================================================================.venv/lib/python3.8/site-packages/pytest_asyncio/plugin.py:39  /home/tyler/edgedb_connect/.venv/lib/python3.8/site-packages/pytest_asyncio/plugin.py:39: PytestDeprecationWarning: direct construction of Function has been deprecated, please use Function.from_parent    item = pytest.Function(name, parent=collector).venv/lib/python3.8/site-packages/pytest_asyncio/plugin.py:45  /home/tyler/edgedb_connect/.venv/lib/python3.8/site-packages/pytest_asyncio/plugin.py:45: PytestDeprecationWarning: direct construction of Function has been deprecated, please use Function.from_parent    item = pytest.Function(name, parent=collector)  # To reload keywords.-- Docs: https://docs.pytest.org/en/latest/warnings.html=========================================================== 3 passed, 2 warnings in 1.71s ============================================================

That'salso barely different. True, but the use of thepytest-asyncio provided decorator is required. If you don't install it, pytest will let you know that you should have and skip the test. You may see some warning from pytest about 'direct construction of Function has been deprecated' It's not your fault,pytest-asyncio needs to accommodate differences in Python's asycio API from version 3.5 to version 3.8 and some of the internals of asyncio have changed during that time. You can safely ignore this warning.

Let's make a pool

The pooled interface is really cool. It allows you to create and allocate async connections to a database without having to re-establish each time.
Let's implement that:

asyncdefconnect_async_pool(self,connection:typing.Optional[EdgeDBConnection]=None,)->edgedb.AsyncIOConnection:ifnotself.pool:self.pool=awaitedgedb.create_async_pool(dsn=self.dsn,host=self.host,port=self.port,admin=bool(self.admin),user=self.user,password=self.password,database=self.database,timeout=self.timeout,min_size=self.pool_min_size,max_size=self.pool_max_size,)returnawaitself.pool.acquire()

This function need a couple more parameters, so we'll have to add those to the dataclass as well:

timeout:int=60pool:typing.Optional[edgedb.AsyncIOPool]=Nonepool_min_size:int=1pool_max_size:int=1

And let's test that that interface works.

@pytest.mark.usefixtures("connection_object")@pytest.mark.asyncioasyncdeftest_edgedb_async_pool(connection_object)->None:async_pool=awaitconnection_object("POOL")assertisinstance(async_pool,edgedb.AsyncIOConnection)awaitasync_pool.aclose()

Let's also silence those warnings from pytest.

(.venv) @agritheory:~/edgedb_connect$ python -m pytest test.py -p no:warnings================================================================ test session starts =================================================================platform linux -- Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1rootdir: /home/tyler/edgedb_connectplugins: asyncio-0.10.0collected 4 items                                                                                                                                    test.py ....                                                                                                                                   [100%]================================================================= 4 passed in 2.45s ==================================================================

An Await Agnostic Interface

Great! We can now connect in several different ways from the same object. But this could still be improved. What if we wanted to store the same connection type (sync/async/pool) and connect that way each time by default?
Well, we know our connection options, so let's put those in a tuple and add a preference to our dataclass:

CONNECTION_TYPES=('SYNC','ASYNC','POOL')# and in the dataclasspool_max_size:int=1connection_type:str='ASYNC'

So how are we going to do this? We can use theEdgeDBConnection object's__call__ method and return the preferred connection type from there.

def__call__(self,connection_type:str="ASYNC")->typing.Union[edgedb.BlockingIOConnection,typing.Coroutine[typing.Any,typing.Any,edgedb.AsyncIOConnection],]:ifnotconnection_type:connection_type=self.connection_typeifconnection_typenotinCONNECTION_TYPES:raiseTypeError(f"'connection_type' must be one of 'SYNC', 'ASYNC' or 'POOL'.\              You provided '{connection_type}'")ifconnection_type=="ASYNC":returnself.connect_async()elifconnection_type=="SYNC":returnself.connect_sync()elifconnection_type=="POOL":returnself.connect_async_pool()

Included is a validation forconnection_type which enforces we don't do something like pass in 'sync' instead of 'SYNC'. Ask me how I know.

@pytest.mark.usefixtures("connection_object")@pytest.mark.xfaildeftest_edgedb_connection_type_validator(connection_object)->typing.NoReturn:sync_connection=connection_object("sync")
(.venv) @agritheory:~/edgedb_connect$ python -m pytest test.py -p no:warnings================================================================ test session starts =================================================================platform linux -- Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1rootdir: /home/tyler/edgedb_connectplugins: asyncio-0.10.0collected 5 items                                                                                                                                    test.py ....x                                                                                                                                  [100%]============================================================ 4 passed, 1 xfailed in 2.48s ============================================================

So let's use this failing example to refactor our other tests. And also add a test for the default condition.

# in test_edgedb_sync_connectionsync_connection=connection_object("SYNC")# in test_edgedb_async_connectionsasync_connection=awaitconnection_object("ASYNC")# in test_edgedb_async_poolasync_pool=awaitconnection_object("POOL")@pytest.mark.usefixtures("connection_object")@pytest.mark.asyncioasyncdeftest_edgedb_default_connection(connection_object)->None:assertconnection_object.connection_type=='ASYNC'default_connection=awaitconnection_object()assertisinstance(default_connection,edgedb.AsyncIOConnection)awaitdefault_connection.aclose()assertdefault_connection.is_closed()isTrue

And test:

(.venv) @agritheory:~/edgedb_connect$ python -m pytest test.py -p no:warnings================================================================ test session starts =================================================================platform linux -- Python 3.8.2, pytest-5.4.1, py-1.8.1, pluggy-0.13.1rootdir: /home/tyler/edgedb_connectplugins: asyncio-0.10.0collected 6 items                                                                                                                                    test.py .....x                                                                                                                                 [100%]============================================================ 5 passed, 1 xfailed in 3.13s ============================================================

But why is that useful?

Fair question. If you were to integrate theedgedb library into an application like Quart or Starlette, you might want to establish a connection and load some of the application's state in an intentionally blocking way and then switch to a non-blocking pattern later on. You could set the default to'ASYNC' or'POOL' but do that initial loading by passing'SYNC' to the connection instance. Things that are running in an event loop still needawait in front of them.

This isn't the end

Honestly, this is one of the least interesting aspects of EdgeDB. But maybe this is interesting enough for you to go out and look atEdgeDB's features, like it's killer schema, built in validations or that it will natively serve you GraphQL.

If you'd like to look at this code in it's finished form,it's here.

Top comments(2)

Subscribe
pic
Create template

Templates let you quickly answer FAQs or store snippets for re-use.

Dismiss
CollapseExpand
 
dmgolembiowski profile image
David M. Golembiowski
  • Joined

Nicely done!

CollapseExpand
 
alescgithub profile image
Alesc
  • Joined

thank you very much, and I hope to find more articles on edgedb. It's very interesting

Are you sure you want to hide this comment? It will become hidden in your post, but will still be visible via the comment'spermalink.

For further actions, you may consider blocking this person and/orreporting abuse

  • Location
    New Hampshire
  • Work
    Principal at AgriTheory
  • Joined

Trending onDEV CommunityHot

DEV Community

We're a place where coders share, stay up-to-date and grow their careers.

Log in Create account

[8]ページ先頭

©2009-2025 Movatter.jp