alex / alchimia Goto Github PK

View Code? Open in Web Editor NEW

102.0 102.0 22.0 93 KB

License: MIT License

Python 100.00%

alchimia's People

Contributors

Stargazers

Watchers

Forkers

jerith emonty dpnova boyombo iffy pombredanne hipchat bjoeris hawkowl evilaliv3 brighid adamchainz glyph j14r hustxujinkang markrwilliams ryban zanachka 0x000000000019b6

alchimia's Issues

add del everywhere

Pretty much every object in alchimia.engine manages some state:

TwistedEngine has a worker for connect & execute
TwistedConnection has a worker where all its SQL is executed
TwistedTransaction holds transaction state - this one we might not need to deal with since it's bound to the connection
TwistedResultProxy holds a result proxy.

TwistedEngine and TwistedConnection are particularly problematic because the worker thread created behind the scenes is its own strong-reference root and will never be cleaned up during the lifetime of the process if these objects do not shut them down.

As seen in #30 , leaking these types of resources can give one the impression that the whole system is just freezing for no reason.

drop python 3.4 since Twisted no longer supports it

The build is now failing:

https://travis-ci.org/alex/alchimia/jobs/568553145

release 0.7

Picking up where #28 left off…

Autoloading tables

Currently, autoloading tables doesn't really work. If I try to do:

table = Table(table_name, metadata, autoload=True)

The call fails because Table._autoload() tries to use Engine.bind_callable(), which isn't defined on TwistedEngine. However, you can do the following:

table = Table(table_name, metadata, autoload=True, autoload_with=engine._engine)

However, it's a bit messy to try to access engine._engine. Would it make sense to either override Engine.bind_callable() or to expose a more formal way of accessing the underlying engine with a property or something?

I believe it's fine to make the blocking call for this use case, since you should only be reflecting on the tables at startup time.

No threads when using MySQL?

I'm writing app to consume event feed and store received events into mysql. Events should be routed to different tables based on a key in them and stored as soon as possible. But whatever i do, i can't get it to spawn more threads and performance is really bad for sequential inserts (~20 inserts/s).

multiple calls (params) => single thread, slow:

for item in data:
    d = engine.execute(table.insert().values(**item))
    d.addCallback(row_handler)

single call (multiparams) => single thread, fast:

d = engine.execute(table.insert(), data)
d.addCallback(all_handler)

I could possibly queue each event type and call multiparams style execute at regular intervals, but i'd rather use separate execute call for each event as soon as it arrives

I'm using version 0.7.0, mysql server 5.7, tested with pymysql, cymysql, mysqlconnector and mysqlclient, all the same.
When using adbapi, threads are created correctly

Am i missing something? Any ideas?

give glyph the ability to release 0.7

#27 is a pretty substantial fix, allowing the correct use of transactions for the first time. Someone (me) should do a release.

In-memory SQLite forgets DDL transaction

I've got some code that creates a SQLite DB, populates a schema, then writes some data.

If the SQLite DB is a file, this seems to work as expected:

2018-06-08T16:10:40-0700 [__main__.Index#info] Importing transmissions...
2018-06-08T16:10:40-0700 [__main__.Index#info] Initializing SQLite DB...
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,028 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,029 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,029 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,029 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,030 INFO sqlalchemy.engine.base.Engine PRAGMA table_info("TRANSMISSION")
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,030 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,031 INFO sqlalchemy.engine.base.Engine INSERT INTO "TRANSMISSION" ("STATION", "SYSTEM", "CHANNEL") VALUES (?, ?, ?)
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,031 INFO sqlalchemy.engine.base.Engine (('Radio 1', 'A', 'Ops'), ('Radio 4', 'B', 'Talk'))
2018-06-08T16:10:40-0700 [stdout#info] 2018-06-08 16:10:40,032 INFO sqlalchemy.engine.base.Engine COMMIT
2018-06-08T16:10:40-0700 [-] Main loop terminated.

However, if the SQLite DB is an in-memory DB, it fails, complaining that the table being written to doesn't exist:

2018-06-08T16:12:34-0700 [__main__.Index#info] Importing transmissions...
2018-06-08T16:12:34-0700 [__main__.Index#info] Initializing SQLite DB...
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,912 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,912 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,913 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,913 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,914 INFO sqlalchemy.engine.base.Engine PRAGMA table_info("TRANSMISSION")
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,914 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,914 INFO sqlalchemy.engine.base.Engine 
2018-06-08T16:12:34-0700 [stdout#info] CREATE TABLE "TRANSMISSION" (
2018-06-08T16:12:34-0700 [stdout#info] 	"STATION" VARCHAR NOT NULL, 
2018-06-08T16:12:34-0700 [stdout#info] 	"SYSTEM" VARCHAR NOT NULL, 
2018-06-08T16:12:34-0700 [stdout#info] 	"CHANNEL" VARCHAR NOT NULL
2018-06-08T16:12:34-0700 [stdout#info] )
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,915 INFO sqlalchemy.engine.base.Engine ()
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,915 INFO sqlalchemy.engine.base.Engine COMMIT
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,916 INFO sqlalchemy.engine.base.Engine INSERT INTO "TRANSMISSION" ("STATION", "SYSTEM", "CHANNEL") VALUES (?, ?, ?)
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,916 INFO sqlalchemy.engine.base.Engine (('Radio 1', 'A', 'Ops'), ('Radio 4', 'B', 'Talk'))
2018-06-08T16:12:34-0700 [stdout#info] 2018-06-08 16:12:34,916 INFO sqlalchemy.engine.base.Engine ROLLBACK
2018-06-08T16:12:34-0700 [-] main function encountered error
	Traceback (most recent call last):
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/internet/defer.py", line 500, in errback
	    self._startRunCallbacks(fail)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/internet/defer.py", line 567, in _startRunCallbacks
	    self._runCallbacks()
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/internet/defer.py", line 653, in _runCallbacks
	    current.result = callback(current.result, *args, **kw)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/internet/defer.py", line 1442, in gotResult
	    _inlineCallbacks(r, g, deferred)
	--- <exception caught here> ---
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/internet/defer.py", line 1384, in _inlineCallbacks
	    result = result.throwExceptionIntoGenerator(g)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/twisted/python/failure.py", line 422, in throwExceptionIntoGenerator
	    return g.throw(self.type, self.value, self.tb)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/mixer/alch_test.py", line 94, in addTransmissions
	    for transmission in transmissions
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/alchimia/engine.py", line 33, in container
	    result = work(*args, **kwargs)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 2075, in execute
	    return connection.execute(statement, *multiparams, **params)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 948, in execute
	    return meth(self, multiparams, params)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/sql/elements.py", line 269, in _execute_on_connection
	    return connection._execute_clauseelement(self, multiparams, params)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1060, in _execute_clauseelement
	    compiled_sql, distilled_params
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1200, in _execute_context
	    context)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1413, in _handle_dbapi_exception
	    exc_info
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 203, in raise_from_cause
	    reraise(type(exception), exception, tb=exc_tb, cause=cause)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/util/compat.py", line 186, in reraise
	    raise value.with_traceback(tb)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/base.py", line 1170, in _execute_context
	    context)
	  File "/Volumes/data/Users/wsanchez/Documents/Developer/BurningMan/radio-mixer/.tox/coverage-py36/lib/python3.6/site-packages/sqlalchemy/engine/default.py", line 505, in do_executemany
	    cursor.executemany(statement, parameters)
	sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: TRANSMISSION [SQL: 'INSERT INTO "TRANSMISSION" ("STATION", "SYSTEM", "CHANNEL") VALUES (?, ?, ?)'] [parameters: (('Radio 1', 'A', 'Ops'), ('Radio 4', 'B', 'Talk'))] (Background on this error at: http://sqlalche.me/e/e3q8)
	
2018-06-08T16:12:34-0700 [-] Main loop terminated.

The difference between these two runs is whether line 71 or line 72 is commented out.

Declarative approach and Sessions. Will be possible to use them?

Hi, great work!

I'm kind of new to both SQLAlchemy and Twisted, but to what I got used to is called Declarative
http://docs.sqlalchemy.org/en/rel_0_9/orm/extensions/declarative.html and as far I know the way to manage this is by using Sessions.

Working with them gives me

'TwistedEngine' object has no attribute 'contextual_connect'

Include public-API-test doubles

Hello!

It would be quite helpful to explicitly confirm whether alchimia.tests.doubles is public or not. If it isn't, both things in there right now are pretty critical for testing apps using alchimia.

I've already filed https://twistedmatrix.com/trac/ticket/8891 upstream.

Motivation?

I would love to see the motivation for this project somewhere? what does it enable me to do?

I could guess, but that's just wrong.

support "threadsafety=1" drivers

Trying to use Alchimia with sqlite3, as documented on the front page, I get:

    sqlalchemy.exc.ProgrammingError: (sqlite3.ProgrammingError) SQLite objects created in a thread can only be used in that same thread.The object was created in thread id 123145310715904 and this is thread id 123145306509312

As documented here https://www.python.org/dev/peps/pep-0249/#threadsafety in the DB-API PEP, the threadsafety attribute of the sqlite3 module is 1, which means you aren't allowed to move connections between threads.

native asynchronous driver support

I'm starting this as a place to have the conversation about a txpostgres implementation of this. I'm willing to work on it as I'm starting a new project that would benefit from it.

Having checked the source so far, I think the majority of the work would be abstracting the deferToThread stuff into a less thread-y naming convention and then hooking up the txpostgres connection objects.

the txpostgres pool is pretty limited compared to the sqla ones, so there may be a little bit of integration worth doing there too.

Anyone have any other thoughts?

Add a pool for TwistedConnection so that we re-use background threads

I keep getting this error when I make many simultaneous queries (klein -> alchimia -> PostgreSQL):

QueuePool limit of size 5 overflow 10 reached, connection timed out, timeout 30

I'm working around it right now by setting max_overflow=-1 (http://docs.sqlalchemy.org/en/rel_0_8/core/pooling.html#sqlalchemy.pool.QueuePool.params.max_overflow) but I think that will cause problems if I actually open more connections than the Postgres database allows. So I think I (or someone) could implement a Deferred-returning pool that returns Connections asynchronously and has requests for connections wait in a queue.

I'm filing this here, because I think alchimia should come with such a pool built in. I'll take a look at what an implementation might look like on Monday.

inserted_primary_key not available

The result of inserting a row doesn't have an inserted_primary_key attribute. I'm thinking about how to do this and will submit a PR if I come up with a solution (or do you already have one?)

0.7.0 broken with Pypy? No idea how to troubleshoot.

Using code substantially similar to @hawkowl (https://atleastfornow.net/blog/alchimia/),
Pypy PyPy 5.4.1
Twisted 16.5.0
alchimia 0.70
psycopg2cffi 2.7.5

QueuePool is exhausted immediately. As far as I can tell connections are never garbage collected, rather they timeout after 30 seconds. The problem is not present when using alchimia 0.6.1 under the same Pypy or running 0.7.0 in under CPython 2.7.12 .

I was unable to distill the failure case to a simpler one. The workload is update heavy. I uploaded the failing code here: https://github.com/nocko/c3next

I am at a loss for how to troubleshoot this. Please advise.

Possibly related to #30?

Engine._execute_clauseelement, &c.

Currently, if you do something like statement.execute() instead of engine.execute(statement), it doesn't work, because Engine._execute_clauseelement isn't overridden.

What do you think of overriding it to just do the same thing as Engine.execute? It looks like it's pretty similar under the hood:

https://github.com/zzzeek/sqlalchemy/blob/f7b119705aef79e7e8957ae6eb654b9057cea45f/lib/sqlalchemy/engine/base.py#L1827

On upgrade to 0.7.0 all calls to engine.execute() are hanging

Python 2.7.5
Twisted==16.4.1
alchimia==0.7.0
psycopg2==2.6.2
SQLAlchemy==1.0.15
Linux ahost-02 3.10.0-229.7.2.el7.x86_64 #1 SMP Tue Jun 23 05:02:16 PDT 2015 x86_64 x86_64 x86_64 GNU/Linux

In my application calling engine.execute() results in hang - Deferred is never called. I injected verbose logging in alchimia/engine.py and it appears the inner function "container" in _defer_to_worker is not being called:

2016-10-05T05:58:43Z [DEBUG] a.e.TwistedEngine - execute called. args=(<sqlalchemy.sql.selectable.Select at 0x4b267d0; Select object>,) kwargs={}
2016-10-05T05:58:43Z [DEBUG] a.e.TwistedEngine - _defer_to_engine called. a=(<sqlalchemy.sql.selectable.Select at 0x4b267d0; Select object>,) f=<bound method Engine.execute of Engine(postgresql+psycopg2://zzz:***@localhost/zzz)> k={}
2016-10-05T05:58:43Z [DEBUG] a.engine - _defer_to_worker called. args=(<sqlalchemy.sql.selectable.Select at 0x4b267d0; Select object>,) worker=<twisted._threads._threadworker.ThreadWorker object at 0x499e290> work=<bound method Engine.execute of Engine(postgresql+psycopg2://zzz:***@localhost/zzz)> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x3224fd0>> kwargs={}

If I change related code to call to engine.connect() the call succeeds as usual. Injected logging:

2016-10-05T06:16:30Z [DEBUG] d.c.FooController - listing foo
2016-10-05T06:16:30Z [DEBUG] a.e.TwistedEngine - connect: called.
2016-10-05T06:16:30Z [DEBUG] a.engine - _threaded_worker#_start_worker: starting thread. target=<function work at 0x50edb18>
2016-10-05T06:16:30Z [DEBUG] a.engine - _threaded_worker#_start_worker: daemonized thead ... starting. thread=<Thread(Thread-3, initial daemon)>
2016-10-05T06:16:30Z [DEBUG] a.e.TwistedEngine - connect: deferring to worker. worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10>
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker called. args=() worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10> work=<bound method Engine.connect of Engine(postgresql+psycopg2://zzz:***@localhost/zzz)> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x33293d0>> kwargs={}
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: running.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: got result. result=<sqlalchemy.engine.base.Connection object at 0x50f67d0>
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: delivered result.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker called. args=() worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10> work=<bound method Connection.begin of <sqlalchemy.engine.base.Connection object at 0x50f67d0>> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x33293d0>> kwargs={}
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: running.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: got result. result=<sqlalchemy.engine.base.RootTransaction object at 0x50e4b50>
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: delivered result.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker called. args=(<sqlalchemy.sql.selectable.Select at 0x50f6810; Select object>,) worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10> work=<bound method Connection.execute of <sqlalchemy.engine.base.Connection object at 0x50f67d0>> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x33293d0>> kwargs={}
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: running.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: got result. result=<sqlalchemy.engine.result.ResultProxy object at 0x50f64d0>
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: delivered result.
2016-10-05T06:16:30Z [DEBUG] d.datastore - got result. result=<alchimia.engine.TwistedResultProxy object at 0x50f68d0>
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker called. args=() worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10> work=<bound method ResultProxy.fetchall of <sqlalchemy.engine.result.ResultProxy object at 0x50f64d0>> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x33293d0>> kwargs={}
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: running.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: got result. result=[…<snip>…]
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: delivered result.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker called. args=() worker=<twisted._threads._threadworker.ThreadWorker object at 0x50e4f10> work=<bound method RootTransaction.commit of <sqlalchemy.engine.base.RootTransaction object at 0x50e4b50>> deliver=<bound method EPollReactor.callFromThread of <twisted.internet.epollreactor.EPollReactor object at 0x33293d0>> kwargs={}
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: running.
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: got result. result=None
2016-10-05T06:16:30Z [DEBUG] a.engine - _defer_to_worker#container: delivered result.
2016-10-05T06:16:30Z [INFO] t.p.log - "127.0.0.1" - - [05/Oct/2016:06:16:29 +0000] "GET /foo HTTP/1.1" 200 1478 "-" "-".

The code in question that hangs:

     def get_foo(self):
        foo = tables.foo
        stmt = select([foo.c.foo_id, foo.c.modified])
        stmt = stmt.order_by(foo.c.foo_id.asc())
        d = self.engine.execute(stmt)
        return d.addCallback(blah) # blah just prints result ... but this is never called back

feature support table?

Do you have a table of sqlalchemy.core features vs features supported by alchimia?