Here is a complete proof of concept using your query, I hope this helps!
from sqlalchemy import *
from sqlalchemy.orm import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import ClauseElement, Executable
class CreateTableAs(Executable, ClauseElement):
def __init__(self, name, query):
self.name = name
if hasattr(query, "statement"):
query = query.statement
self.query = query
@compiles(CreateTableAs, "postgresql")
def _create_table_as(element, compiler, **kw):
return "CREATE TABLE %s AS %s" % (
element.name,
compiler.process(element.query),
)
Base = declarative_base()
class Action(Base):
__tablename__ = "action"
id = Column(Integer, primary_key=True)
some_other_data = Column(String)
statement1 = Column(String)
statement2 = Column(String)
name = Column(String)
source_file = Column(String)
e = create_engine("postgresql://scott:tiger@localhost/test", echo=True)
Base.metadata.drop_all(e)
Base.metadata.create_all(e)
s = session = Session(e)
s.add_all(
[
Action(
some_other_data="some other data %d" % i,
statement1="s1 %d" % (i % 4),
statement2="s2 %d" % (i % 2),
name="name %d" % (i % 4),
source_file="some file %d" % (i % 4),
)
for i in range(10)
]
)
session.flush()
q = (
session.query(
Action.statement1,
Action.statement2,
Action.name,
Action.source_file,
func.count("*").label("C"),
)
.group_by(
Action.statement1, Action.statement2, Action.name, Action.source_file
)
.having(func.count("*") > 1)
)
session.execute(CreateTableAs("t2", q))
class T2(Base):
__table__ = Table("t2", Base.metadata, autoload_with=session.connection())
__mapper_args__ = {
"primary_key": [__table__.c.source_file] # a primary key must
be determined
}
print(
session.query(T2).all()
)
On Thu, Apr 18, 2019 at 10:27 AM Markus Elfring <[email protected]> wrote:
>
> > can you perhaps illustrate a code example of what you would like to do ?
>
> I have tried the following approach out for the software “SQLAlchemy 1.3.2”
> together with the engine “sqlite:///:memory:”.
>
> …
> q = session.query(action.statement1, action.statement2, action.name,
> action.source_file,
> func.count("*").label("C")
> ).group_by(action.statement1,
> action.statement2,
> action.name,
> action.source_file) \
> .having(func.count("*") > 1)
> ct = 'create table t2 as ' + str(q.statement)
> sys.stderr.write("Command: " + ct + "\n")
> session.execute(ct)
> …
>
>
> Now I wonder about parameter specifications in the generated SQL command
> (and a traceback with the corresponding error code
> “http://sqlalche.me/e/cd3x”).
>
> Command: create table t2 as SELECT …, statements.source_file, count(:count_1)
> AS "C"
> FROM … GROUP BY …
> HAVING count(:count_2) > :count_3
>
>
> Which details should be adjusted a bit more here?
>
> Regards,
> Markus
--
SQLAlchemy -
The Python SQL Toolkit and Object Relational Mapper
http://www.sqlalchemy.org/
To post example code, please provide an MCVE: Minimal, Complete, and Verifiable
Example. See http://stackoverflow.com/help/mcve for a full description.
---
You received this message because you are subscribed to the Google Groups
"sqlalchemy" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/sqlalchemy.
For more options, visit https://groups.google.com/d/optout.
