diff --git a/sqlx/src/lib.rs b/sqlx/src/lib.rs
index afade9e8..a501beed 100644
--- a/sqlx/src/lib.rs
+++ b/sqlx/src/lib.rs
@@ -46,6 +46,7 @@
 #[cfg(feature = "blocking")]
 pub mod blocking;
 
+mod query;
 mod runtime;
 
 #[cfg(feature = "mysql")]
@@ -56,6 +57,7 @@ pub mod mysql;
 
 #[cfg(feature = "blocking")]
 pub use blocking::Blocking;
+pub use query::{query, Query};
 pub use runtime::DefaultRuntime;
 #[cfg(feature = "actix")]
 pub use sqlx_core::Actix;
@@ -66,6 +68,6 @@ pub use sqlx_core::AsyncStd;
 #[cfg(feature = "tokio")]
 pub use sqlx_core::Tokio;
 pub use sqlx_core::{
-    Acquire, Close, Connect, ConnectOptions, Connection, Database, Error, Executor, Result, Row,
-    Runtime,
+    Acquire, Arguments, Close, Connect, ConnectOptions, Connection, Database, Error, Executor,
+    Result, Row, Runtime,
 };
diff --git a/sqlx/src/mysql/connection.rs b/sqlx/src/mysql/connection.rs
index 5924250e..ec562dd1 100644
--- a/sqlx/src/mysql/connection.rs
+++ b/sqlx/src/mysql/connection.rs
@@ -3,14 +3,14 @@ use std::ops::{Deref, DerefMut};
 
 #[cfg(feature = "async")]
 use futures_util::future::{BoxFuture, FutureExt};
-use sqlx_core::Executor;
+use sqlx_core::{Execute, Executor};
 
 use super::{MySql, MySqlConnectOptions, MySqlQueryResult, MySqlRow};
 #[cfg(feature = "blocking")]
 use crate::blocking;
+use crate::{Arguments, Close, Connect, Connection, DefaultRuntime, Runtime};
 #[cfg(feature = "async")]
 use crate::{Async, Result};
-use crate::{Close, Connect, Connection, DefaultRuntime, Runtime};
 
 /// A single connection (also known as a session) to a MySQL database server.
 #[allow(clippy::module_name_repetitions)]
@@ -50,20 +50,32 @@ impl<Rt: Async> MySqlConnection<Rt> {
 
     // TODO: document from Executor
 
-    pub async fn execute(&mut self, sql: &str) -> Result<MySqlQueryResult> {
-        self.0.execute(sql).await
+    pub async fn execute<'q, 'a, E>(&mut self, query: E) -> Result<MySqlQueryResult>
+    where
+        E: Execute<'q, 'a, MySql>,
+    {
+        self.0.execute(query).await
     }
 
-    pub async fn fetch_all(&mut self, sql: &str) -> Result<Vec<MySqlRow>> {
-        self.0.fetch_all(sql).await
+    pub async fn fetch_all<'q, 'a, E>(&mut self, query: E) -> Result<Vec<MySqlRow>>
+    where
+        E: Execute<'q, 'a, MySql>,
+    {
+        self.0.fetch_all(query).await
     }
 
-    pub async fn fetch_one(&mut self, sql: &str) -> Result<MySqlRow> {
-        self.0.fetch_one(sql).await
+    pub async fn fetch_one<'q, 'a, E>(&mut self, query: E) -> Result<MySqlRow>
+    where
+        E: Execute<'q, 'a, MySql>,
+    {
+        self.0.fetch_one(query).await
     }
 
-    pub async fn fetch_optional(&mut self, sql: &str) -> Result<Option<MySqlRow>> {
-        self.0.fetch_optional(sql).await
+    pub async fn fetch_optional<'q, 'a, E>(&mut self, query: E) -> Result<Option<MySqlRow>>
+    where
+        E: Execute<'q, 'a, MySql>,
+    {
+        self.0.fetch_optional(query).await
     }
 
     /// Explicitly close this database connection.
@@ -125,34 +137,42 @@ impl<Rt: Runtime> Executor<Rt> for MySqlConnection<Rt> {
     type Database = MySql;
 
     #[cfg(feature = "async")]
-    fn execute<'x, 'e, 'q>(&'e mut self, sql: &'q str) -> BoxFuture<'x, Result<MySqlQueryResult>>
+    fn execute<'x, 'e, 'q, 'a, E>(&'e mut self, query: E) -> BoxFuture<'x, Result<MySqlQueryResult>>
     where
         Rt: Async,
+        E: 'x + Execute<'q, 'a, MySql>,
         'e: 'x,
         'q: 'x,
+        'a: 'x,
     {
-        self.0.execute(sql)
+        self.0.execute(query)
     }
 
-    fn fetch_all<'x, 'e, 'q>(&'e mut self, sql: &'q str) -> BoxFuture<'x, Result<Vec<MySqlRow>>>
+    #[cfg(feature = "async")]
+    fn fetch_all<'x, 'e, 'q, 'a, E>(&'e mut self, query: E) -> BoxFuture<'x, Result<Vec<MySqlRow>>>
     where
         Rt: Async,
+        E: 'x + Execute<'q, 'a, MySql>,
         'e: 'x,
         'q: 'x,
+        'a: 'x,
     {
-        self.0.fetch_all(sql)
+        self.0.fetch_all(query)
     }
 
-    fn fetch_optional<'x, 'e, 'q>(
+    #[cfg(feature = "async")]
+    fn fetch_optional<'x, 'e, 'q, 'a, E>(
         &'e mut self,
-        sql: &'q str,
+        query: E,
     ) -> BoxFuture<'x, Result<Option<MySqlRow>>>
     where
         Rt: Async,
+        E: 'x + Execute<'q, 'a, MySql>,
         'e: 'x,
         'q: 'x,
+        'a: 'x,
     {
-        self.0.fetch_optional(sql)
+        self.0.fetch_optional(query)
     }
 }
 
diff --git a/sqlx/src/query.rs b/sqlx/src/query.rs
new file mode 100644
index 00000000..f98f4fc8
--- /dev/null
+++ b/sqlx/src/query.rs
@@ -0,0 +1,66 @@
+use std::borrow::Cow;
+use std::marker::PhantomData;
+
+use sqlx_core::{Execute, Executor, TypeEncode};
+
+use crate::{Arguments, Database, DefaultRuntime, Runtime};
+
+pub struct Query<'q, 'a, Db: Database, Rt: Runtime = DefaultRuntime> {
+    sql: Cow<'q, str>,
+    arguments: Arguments<'a, Db>,
+    runtime: PhantomData<Rt>,
+}
+
+impl<'q, 'a, Db: Database, Rt: Runtime> Execute<'q, 'a, Db> for Query<'q, 'a, Db, Rt> {
+    fn sql(&self) -> &str {
+        &self.sql
+    }
+
+    fn arguments(&self) -> Option<&Arguments<'a, Db>> {
+        Some(&self.arguments)
+    }
+}
+
+impl<'q, 'a, Db: Database, Rt: Runtime> Query<'q, 'a, Db, Rt> {
+    pub fn bind<T: 'a + TypeEncode<Db>>(&mut self, value: &'a T) -> &mut Self {
+        self.arguments.add(value);
+        self
+    }
+}
+
+#[cfg(feature = "async")]
+impl<'q, 'a, Db: Database, Rt: crate::Async> Query<'q, 'a, Db, Rt> {
+    pub async fn execute<X>(&self, mut executor: X) -> crate::Result<Db::QueryResult>
+    where
+        X: Executor<Rt, Database = Db>,
+    {
+        executor.execute(self).await
+    }
+
+    pub async fn fetch_optional<X>(&self, mut executor: X) -> crate::Result<Option<Db::Row>>
+    where
+        X: Executor<Rt, Database = Db>,
+    {
+        executor.fetch_optional(self).await
+    }
+
+    pub async fn fetch_one<X>(&self, mut executor: X) -> crate::Result<Db::Row>
+    where
+        X: Executor<Rt, Database = Db>,
+    {
+        executor.fetch_one(self).await
+    }
+
+    pub async fn fetch_all<X>(&self, mut executor: X) -> crate::Result<Vec<Db::Row>>
+    where
+        X: Executor<Rt, Database = Db>,
+    {
+        executor.fetch_all(self).await
+    }
+}
+
+pub fn query<'q, 'a, Db: Database, Rt: Runtime>(
+    sql: impl Into<Cow<'q, str>>,
+) -> Query<'q, 'a, Db, Rt> {
+    Query::<'q, 'a, Db, Rt> { sql: sql.into(), arguments: Arguments::new(), runtime: PhantomData }
+}