diff --git a/aiomysql/connection.py b/aiomysql/connection.py
index 8bf5cbf4..7f5085b4 100644
--- a/aiomysql/connection.py
+++ b/aiomysql/connection.py
@@ -15,7 +15,6 @@
 from pymysql.constants import SERVER_STATUS
 from pymysql.constants import CLIENT
 from pymysql.constants import COMMAND
-from pymysql.constants import FIELD_TYPE
 from pymysql.util import byte2int, int2byte
 from pymysql.converters import (escape_item, encoders, decoders,
                                 escape_string, escape_bytes_prefixed, through)
@@ -25,7 +24,7 @@
                          IntegrityError, InternalError, NotSupportedError,
                          ProgrammingError)
 
-from pymysql.connections import TEXT_TYPES, MAX_PACKET_LEN, DEFAULT_CHARSET
+from pymysql.connections import MAX_PACKET_LEN, DEFAULT_CHARSET
 from pymysql.connections import _auth
 
 from pymysql.connections import pack_int24
@@ -39,8 +38,9 @@
 
 # from aiomysql.utils import _convert_to_str
 from .cursors import Cursor
-from .utils import _ConnectionContextManager, _ContextManager
+from .utils import _ConnectionContextManager, _ContextManager, _decide_encoding
 from .log import logger
+from .prepared_statement import PreparedStatement
 
 DEFAULT_USER = getpass.getuser()
 
@@ -971,6 +971,38 @@ async def sha256_password_auth(self, pkt):
         pkt.check_error()
         return pkt
 
+    async def prepare(self, sql):
+        await self._execute_command(COMMAND.COM_STMT_PREPARE, sql)
+        packet = await self._read_packet()
+        if not packet.is_ok_packet():
+            raise Error("Unexpected error")
+
+        # status
+        packet.advance(1)
+        statement_id = packet.read_uint32()
+        num_columns = packet.read_uint16()
+        num_params = packet.read_uint16()
+        # reserved
+        packet.advance(1)
+        # warning count
+        packet.read_uint16()
+        params = []
+        columns = []
+        if num_params > 0:
+            for _ in range(num_params):
+                params.append(await self._read_packet(FieldDescriptorPacket))
+            if self.client_flag | CLIENT.PROTOCOL_41:
+                # EOF
+                await self._read_packet()
+        if num_columns > 0:
+            for _ in range(num_columns):
+                columns.append(await self._read_packet(FieldDescriptorPacket))
+            if self.client_flag | CLIENT.PROTOCOL_41:
+                # EOF
+                await self._read_packet()
+
+        return PreparedStatement(self, statement_id, params, columns)
+
     # _mysql support
     def thread_id(self):
         return self.server_thread_id[0]
@@ -1244,30 +1276,8 @@ async def _get_descriptions(self):
                 FieldDescriptorPacket)
             self.fields.append(field)
             description.append(field.description())
-            field_type = field.type_code
-            if use_unicode:
-                if field_type == FIELD_TYPE.JSON:
-                    # When SELECT from JSON column: charset = binary
-                    # When SELECT CAST(... AS JSON): charset = connection
-                    # encoding
-                    # This behavior is different from TEXT / BLOB.
-                    # We should decode result by connection encoding
-                    # regardless charsetnr.
-                    # See https://github.com/PyMySQL/PyMySQL/issues/488
-                    encoding = conn_encoding  # SELECT CAST(... AS JSON)
-                elif field_type in TEXT_TYPES:
-                    if field.charsetnr == 63:  # binary
-                        # TEXTs with charset=binary means BINARY types.
-                        encoding = None
-                    else:
-                        encoding = conn_encoding
-                else:
-                    # Integers, Dates and Times, and other basic data
-                    # is encoded in ascii
-                    encoding = 'ascii'
-            else:
-                encoding = None
-            converter = self.connection.decoders.get(field_type)
+            encoding = _decide_encoding(use_unicode, conn_encoding, field)
+            converter = self.connection.decoders.get(field.type_code)
             if converter is through:
                 converter = None
             self.converters.append((encoding, converter))
diff --git a/aiomysql/cursors.py b/aiomysql/cursors.py
index 97f0431f..c493d75f 100644
--- a/aiomysql/cursors.py
+++ b/aiomysql/cursors.py
@@ -3,13 +3,13 @@
 import warnings
 import contextlib
 
+from pymysql.constants import FIELD_TYPE
 from pymysql.err import (
     Warning, Error, InterfaceError, DataError,
     DatabaseError, OperationalError, IntegrityError, InternalError,
     NotSupportedError, ProgrammingError)
 
 from .log import logger
-from .connection import FIELD_TYPE
 
 # https://github.com/PyMySQL/PyMySQL/blob/master/pymysql/cursors.py#L11-L18
 
diff --git a/aiomysql/prepared_statement.py b/aiomysql/prepared_statement.py
new file mode 100644
index 00000000..3971adb9
--- /dev/null
+++ b/aiomysql/prepared_statement.py
@@ -0,0 +1,294 @@
+import datetime
+from decimal import Decimal
+import struct
+
+from pymysql.connections import (FieldDescriptorPacket, MysqlPacket)
+from pymysql.constants import (COMMAND, FIELD_TYPE)
+from pymysql.converters import through
+from pymysql.err import Error
+
+from .utils import _decide_encoding
+
+
+class PreparedStatement(object):
+    def __init__(self, connection, stmt_id, params, columns):
+        self.connection = connection
+        self.stmt_id = stmt_id
+        self.params = params
+        self.columns = columns
+        self._rows = None
+        self._rownumber = 0
+        self._rowcount = 0
+
+    async def execute(self, *args):
+        if len(args) != len(self.params):
+            raise Error("argument count doesn't match")
+        self.connection._next_seq_id = 0
+        self._rownumber = 0
+        data = struct.pack("!B", COMMAND.COM_STMT_EXECUTE)
+        data += struct.pack("<I", self.stmt_id)
+        # CURSOR_TYPE_NO_CURSOR
+        data += struct.pack("!B", 0)
+        # iteration count, always 1
+        data += struct.pack("<I", 1)
+        data += self._encode_params(args)
+        self.connection.write_packet(data)
+
+        self._rows = await self._read_result()
+        self._rowcount = len(self._rows)
+        return self._rowcount
+
+    def _encode_params(self, args):
+        if len(args) == 0:
+            return b''
+        null_bitmap = [0] * ((len(args) + 7) // 8)
+        new_params_bound_flag = b'\x01'
+        type_data = b''
+        value_data = b''
+        for i, arg in enumerate(args):
+            if arg is None:
+                null_bitmap[i // 8] |= 1 << (i % 8)
+                type_data += bytes([FIELD_TYPE.NULL, 0])
+            elif isinstance(arg, int):
+                type_data += bytes([FIELD_TYPE.LONGLONG, 0])
+                value_data += struct.pack("<q", arg)
+            elif isinstance(arg, float):
+                type_data += bytes([FIELD_TYPE.DOUBLE, 0])
+                value_data += struct.pack("<d", arg)
+            elif isinstance(arg, Decimal):
+                type_data += bytes([FIELD_TYPE.STRING, 0])
+                data = str(arg).encode('utf-8')
+                value_data += \
+                    self._get_encoded_integer_on_length(len(data)) + data
+            elif isinstance(arg, bytes):
+                type_data += bytes([FIELD_TYPE.STRING, 0])
+                value_data += \
+                    self._get_encoded_integer_on_length(len(arg)) + arg
+            elif isinstance(arg, str):
+                type_data += bytes([FIELD_TYPE.STRING, 0])
+                data = arg.encode("utf-8")
+                value_data += \
+                    self._get_encoded_integer_on_length(len(data)) + data
+            elif isinstance(arg, (datetime.date, datetime.datetime)):
+                type_data += bytes([FIELD_TYPE.STRING, 0])
+                data = arg.strftime("%Y-%m-%d %H:%M:%S.%f").encode('utf-8')
+                value_data += \
+                    self._get_encoded_integer_on_length(len(data)) + data
+                continue
+            else:
+                raise Error("unsupported type " + str(type(arg)))
+        return (
+            bytes(null_bitmap) + new_params_bound_flag + type_data + value_data
+        )
+
+    async def fetchone(self):
+        if self._rows is None:
+            return None
+
+        result = self._rows[self._rownumber]
+        self._rownumber += 1
+        return result
+
+    async def fetchmany(self, size=1):
+        if self._rows is None:
+            return []
+
+        end = self._rownumber + size
+        result = self._rows[self._rownumber:end]
+        self._rownumber = min(end, len(self._rows))
+
+        return result
+
+    async def fetchall(self):
+        if self._rows is None:
+            return []
+
+        if self._rownumber:
+            result = self._rows[self._rownumber:]
+        else:
+            result = self._rows
+        self._rownumber = len(self._rows)
+        return result
+
+    async def _read_result(self):
+        # noinspection PyProtectedMember
+        packet = await self.connection._read_packet()
+        if packet.is_ok_packet():
+            return []
+        columns = []
+        columns_num = packet.read_uint8()
+        # read column definitions
+        for _ in range(columns_num):
+            # noinspection PyProtectedMember
+            columns.append(
+                await self.connection._read_packet(FieldDescriptorPacket))
+        # noinspection PyProtectedMember
+        packet = await self.connection._read_packet()
+        if not packet.is_eof_packet():
+            raise Error("expecting EOF packet")
+        converters = []
+        for c in columns:
+            encoding = _decide_encoding(
+                self.connection.use_unicode, self.connection.encoding, c)
+            converter = self.connection.decoders.get(c.type_code)
+            if converter is through:
+                converter = None
+            converters.append((encoding, converter))
+        result = []
+        while True:
+            # noinspection PyProtectedMember
+            packet = await self.connection._read_packet(BinaryResultSetPacket)
+            if packet.is_eof_packet():
+                return result
+            packet.set_columns(columns)
+            packet.set_converters(converters)
+            result.append(packet.get_result())
+
+    def _get_encoded_integer_on_length(self, size):
+        if size <= 250:
+            return bytes([size])
+        elif size <= 0xffff:
+            return b"\xfc" + struct.pack("<H", size)
+        elif size <= 0xffffff:
+            return b"\xfd" + struct.pack("<BH", size & 0xff, size >> 8)
+        elif size <= 0xffffffff:
+            return b"\xfe" + struct.pack("<L", size)
+        raise Error("size too long")
+
+
+_string_types = {
+    FIELD_TYPE.STRING, FIELD_TYPE.VARCHAR, FIELD_TYPE.VAR_STRING,
+    FIELD_TYPE.ENUM, FIELD_TYPE.SET, FIELD_TYPE.LONG_BLOB,
+    FIELD_TYPE.MEDIUM_BLOB, FIELD_TYPE.BLOB, FIELD_TYPE.TINY_BLOB,
+    FIELD_TYPE.GEOMETRY, FIELD_TYPE.BIT, FIELD_TYPE.DECIMAL,
+    FIELD_TYPE.NEWDECIMAL, FIELD_TYPE.JSON,
+}
+_date_types = {
+    FIELD_TYPE.DATE, FIELD_TYPE.DATETIME, FIELD_TYPE.TIMESTAMP,
+}
+
+
+class BinaryResultSetPacket(MysqlPacket):
+    def __init__(self, data, encoding):
+        super().__init__(data, encoding)
+        self._columns = []
+        self._converters = {}
+
+    def get_result(self):
+        if self.is_eof_packet():
+            return []
+        # header
+        self.advance(1)
+        # parse binary result row
+        # https://dev.mysql.com/doc/internals/en/binary-protocol-resultset-row.html
+        null_bitmap_len = (len(self._columns) + 7 + 2) >> 3
+        null_bitmap = self.read(null_bitmap_len)
+        result = []
+        for i, c in enumerate(self._columns):
+            if null_bitmap[(i + 2) >> 3] & (1 << ((i + 2) % 8)):
+                result.append(None)
+                continue
+            # https://dev.mysql.com/doc/internals/en/binary-protocol-value.html
+            if c.type_code in _string_types:
+                n, is_none = self._read_length_encoded_integer()
+                if is_none:
+                    result.append(None)
+                    continue
+                data = self.read(n)
+                encoding, converter = self._converters[i]
+                if encoding is not None:
+                    data = data.decode(encoding)
+                if converter is not None:
+                    data = converter(data)
+                result.append(data)
+                continue
+            if c.type_code == FIELD_TYPE.LONGLONG:
+                result.append(self.read_struct("<q")[0])
+                continue
+            if c.type_code in {FIELD_TYPE.LONG, FIELD_TYPE.INT24}:
+                result.append(self.read_struct("<i")[0])
+                continue
+            if c.type_code in {FIELD_TYPE.SHORT, FIELD_TYPE.YEAR}:
+                result.append(self.read_struct("<h")[0])
+                continue
+            if c.type_code == FIELD_TYPE.TINY:
+                result.append(self.read_struct("<b")[0])
+                continue
+            if c.type_code == FIELD_TYPE.DOUBLE:
+                result.append(self.read_struct("<d")[0])
+                continue
+            if c.type_code == FIELD_TYPE.FLOAT:
+                result.append(self.read_struct("<f")[0])
+                continue
+            if c.type_code in _date_types:
+                n = self.read_uint8()
+                if n == 0:
+                    result.append(None)
+                    continue
+                if n not in {4, 7, 11}:
+                    raise Error("unexpected data")
+                year = self.read_uint16()
+                month = self.read_uint8()
+                day = self.read_uint8()
+                if n == 4:
+                    result.append(datetime.date(year, month, day))
+                    continue
+                hour = self.read_uint8()
+                minute = self.read_uint8()
+                second = self.read_uint8()
+                if n == 7:
+                    result.append(
+                        datetime.datetime(
+                            year, month, day, hour, minute, second))
+                    continue
+                microsecond = self.read_uint32()
+                if n == 11:
+                    result.append(
+                        datetime.datetime(
+                            year, month, day, hour, minute, second, microsecond
+                        ))
+                    continue
+            if c.type_code == FIELD_TYPE.TIME:
+                n = self.read_uint8()
+                if n == 0:
+                    result.append(None)
+                    continue
+                if n not in {8, 12}:
+                    raise Error("unexpected data")
+                negate = -1 if self.read_uint8() == 1 else 1
+                days = self.read_uint32()
+                hour = self.read_uint8()
+                minute = self.read_uint8()
+                second = self.read_uint8()
+                time_delta = datetime.timedelta(
+                    days=days, seconds=second, minutes=minute, hours=hour
+                ) * negate
+                if n == 8:
+                    result.append(time_delta)
+                    continue
+                time_delta += \
+                    datetime.timedelta(microseconds=self.read_uint32())
+                result.append(time_delta)
+                continue
+
+        return tuple(result)
+
+    def set_columns(self, columns):
+        self._columns = columns
+
+    def set_converters(self, converters):
+        self._converters = converters
+
+    def _read_length_encoded_integer(self):
+        n = self.read_uint8()
+        if n < 0xfb:
+            return n, False
+        if n == 0xfb:
+            return 0, True
+        if n == 0xfc:
+            return self.read_uint16(), False
+        if n == 0xfd:
+            return self.read_uint24(), False
+        if n == 0xfe:
+            return self.read_uint64(), False
+        raise Error("unexpected value")
diff --git a/aiomysql/utils.py b/aiomysql/utils.py
index 9b74b9df..f37f558b 100644
--- a/aiomysql/utils.py
+++ b/aiomysql/utils.py
@@ -1,5 +1,8 @@
 from collections.abc import Coroutine
 
+from pymysql.connections import TEXT_TYPES
+from pymysql.constants import FIELD_TYPE
+
 
 class _ContextManager(Coroutine):
 
@@ -149,3 +152,30 @@ async def __aexit__(self, exc_type, exc_val, exc_tb):
         finally:
             self._pool = None
             self._conn = None
+
+
+def _decide_encoding(use_unicode, conn_encoding, field):
+    field_type = field.type_code
+    if use_unicode:
+        if field_type == FIELD_TYPE.JSON:
+            # When SELECT from JSON column: charset = binary
+            # When SELECT CAST(... AS JSON): charset = connection
+            # encoding
+            # This behavior is different from TEXT / BLOB.
+            # We should decode result by connection encoding
+            # regardless charsetnr.
+            # See https://github.com/PyMySQL/PyMySQL/issues/488
+            encoding = conn_encoding  # SELECT CAST(... AS JSON)
+        elif field_type in TEXT_TYPES:
+            if field.charsetnr == 63:  # binary
+                # TEXTs with charset=binary means BINARY types.
+                encoding = None
+            else:
+                encoding = conn_encoding
+        else:
+            # Integers, Dates and Times, and other basic data
+            # is encoded in ascii
+            encoding = 'ascii'
+    else:
+        encoding = None
+    return encoding
diff --git a/tests/test_prepared_statement.py b/tests/test_prepared_statement.py
new file mode 100644
index 00000000..54dda389
--- /dev/null
+++ b/tests/test_prepared_statement.py
@@ -0,0 +1,102 @@
+from decimal import Decimal
+import datetime
+import json
+
+import pytest
+
+
+async def prepare(conn):
+    c = await conn.cursor()
+
+    # create a table ane some data to query
+    await c.execute("drop table if exists everytype")
+    await c.execute(
+        """CREATE TABLE everytype (
+            colchar char(20),
+            colset set('a', 'b'),
+            colblob blob,
+            colenum enum('a', 'b'),
+            colgeometry geometry,
+            colbit bit,
+            coldecimal decimal,
+            coljson json,
+            collong bigint,
+            colint integer,
+            colshort smallint,
+            coltiny tinyint,
+            coldouble double,
+            colfloat float,
+            coldate date,
+            coldatetime datetime,
+            coltime time,
+            colint2 integer
+        )"""
+    )
+
+
+@pytest.mark.run_loop
+async def test_simple(connection):
+    prepared = await connection.prepare("SELECT 42")
+    await prepared.execute()
+    r = await prepared.fetchone()
+    assert r[0] == 42, "expecting 42"
+
+
+@pytest.mark.run_loop
+async def test(connection):
+    await prepare(connection)
+    prepared = await connection.prepare(
+        """
+        INSERT INTO everytype (
+            colchar,
+            colset,
+            colblob,
+            colenum,
+            colgeometry,
+            colbit,
+            coldecimal,
+            coljson,
+            collong,
+            colint,
+            colshort,
+            coltiny,
+            coldouble,
+            colfloat,
+            coldate,
+            coldatetime,
+            coltime,
+            colint2
+        ) VALUES
+        (?,?,?,?,ST_GeomFromText('POINT(1 1)'),?,?,?,?,?,?,?,?,?,?,?,?,?)
+        """
+    )
+    await prepared.execute(
+        "123", "a", b"\x01\x02", "b", 1, Decimal("123"), '{"a":"b"}', 123, 123,
+        123, 123, 123.45, 123.45, datetime.date(2020, 6, 1),
+        datetime.datetime(2020, 6, 1, 1, 23, 45, 678),
+        datetime.datetime(1, 1, 1, 1, 23, 45), None)
+
+    prepared = await connection.prepare("SELECT * FROM everytype")
+    await prepared.execute()
+    r = await prepared.fetchone()
+    assert r[0] == "123"
+    assert r[1] == "a"
+    assert r[2] == b"\x01\x02"
+    assert r[3] == "b"
+    # geometry
+    assert isinstance(r[4], bytes)
+    with pytest.raises(UnicodeDecodeError):
+        r[4].decode('utf-8')
+    assert r[5] == b'\x01'
+    assert r[6] == Decimal("123")
+    assert json.loads(r[7]) == json.loads('{"a":"b"}')
+    assert r[8] == 123
+    assert r[9] == 123
+    assert r[10] == 123
+    assert r[11] == 123
+    assert pytest.approx(r[12]) == 123.45
+    assert pytest.approx(r[13]) == 123.45
+    assert r[14] == datetime.date(2020, 6, 1)
+    assert r[15] == datetime.datetime(2020, 6, 1, 1, 23, 45)
+    assert r[16] == datetime.timedelta(hours=1, minutes=23, seconds=45)
+    assert r[17] is None